diff --git a/.github/workflows/module-durable.yml b/.github/workflows/module-durable.yml index 36ac2a9c..bb5d3b8a 100644 --- a/.github/workflows/module-durable.yml +++ b/.github/workflows/module-durable.yml @@ -86,6 +86,7 @@ jobs: cargo test -p lyracore-gateway --test startup_relays -- --ignored --test-threads=1 cargo test -p lyracore-gateway --test notification_events -- --ignored --test-threads=1 cargo test -p lyracore-gateway --bin lyracore-gateway stdb::reads::indexed_durable_tests -- --ignored --test-threads=1 + cargo test -p lyracore-gateway --bin lyracore-gateway stdb::subscription_scope_durable_tests -- --ignored --test-threads=1 cargo test -p lyracore-gateway --bin lyracore-gateway stdb::world_view::item_durable_tests::live_worn_item_insert_enchant_and_delete_reach_a_nearby_peer -- --ignored --exact - name: Run Benilla against private Module instances diff --git a/docs/architecture.md b/docs/architecture.md index 90850c6f..0689552e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -477,12 +477,15 @@ verbs go through the module's operator-gated `gw_*` surface with the actor named per-player connection exists anywhere (#483; the account's "bound identity" is now the derived `synthetic_owner_identity`, minted by no connection and presented by no client). -A coordinator connection subscribes a base list of literal `SELECT * FROM ` queries, plus -the sharded tables when more than one database is configured (`coordinator_queries` in -`connection.rs`). The sharded tables are conditional -because **a subscription to a table the deployed module does not have fails to apply**, which would -fail the whole gateway — so a gateway restarted before a module republish must not ask for the -sharded tables. +A Coordinator subscribes literal table queries and the Relay table declarations in +`gateway/src/stdb/connection.rs`. Multi-database configurations also subscribe the Shard control +tables. A dedicated Realm-core connection omits 30 world catalogues, including creatures, spells, +quests and taxi routes. It retains item catalogues needed by Mail and auctions, faction relationships, +and every dynamic, control and Relay table. If Realm-core also hosts a configured World Shard, it +keeps the full query set. Idle World Shards keep their subscriptions. Reconnect preserves the role. + +Publish a Module that contains the required tables before starting this Gateway. An unknown table +causes subscription setup to fail; it does not produce an empty optional cache. ### 5.2 Area of interest — default on diff --git a/gateway/src/stdb/connection.rs b/gateway/src/stdb/connection.rs index dbde1f7f..45ad2604 100644 --- a/gateway/src/stdb/connection.rs +++ b/gateway/src/stdb/connection.rs @@ -660,7 +660,7 @@ pub(crate) struct CoordinatorInner { token: Option, uri: String, db_name: String, - sharded_tables: bool, + subscription: CoordinatorSubscription, call_pipes: Vec, call_pipe_next: std::sync::atomic::AtomicUsize, pub(crate) party_command_lane_overflow: AtomicBool, @@ -978,7 +978,61 @@ fn call_pipe_needs_repair(is_healthy: bool, already_reconnecting: bool) -> bool !is_healthy && !already_reconnecting } -fn coordinator_queries(sharded_tables: bool) -> Vec<&'static str> { +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum CoordinatorSubscription { + SingleShard, + WorldShard, + RealmCore, +} + +impl CoordinatorSubscription { + fn for_shard(map: &ShardMap, shard: &str) -> Self { + if map.databases().len() <= 1 { + Self::SingleShard + } else if map.realm_core_db() == Some(shard) + && !map.shards().iter().any(|world| world == shard) + { + Self::RealmCore + } else { + Self::WorldShard + } + } +} + +const WORLD_CATALOG_QUERIES: &[&str] = &[ + "SELECT * FROM game_creature_template", + "SELECT * FROM game_creature_spawn_map", + "SELECT * FROM game_area", + "SELECT * FROM game_taxi_node", + "SELECT * FROM game_taxi_path", + "SELECT * FROM game_taxi_path_node", + "SELECT * FROM game_start_position", + "SELECT * FROM game_spell", + "SELECT * FROM game_spell_effect", + "SELECT * FROM game_npc_vendor", + "SELECT * FROM game_gossip_menu", + "SELECT * FROM game_gossip_condition", + "SELECT * FROM game_gossip_option", + "SELECT * FROM game_gossip_menu_profile", + "SELECT * FROM game_gossip_menu_profile_option", + "SELECT * FROM game_creature_gossip_menu_override", + "SELECT * FROM game_npc_text", + "SELECT * FROM game_npc_text_slot", + "SELECT * FROM game_quest_template", + "SELECT * FROM game_quest_text", + "SELECT * FROM game_quest_objective", + "SELECT * FROM game_quest_reward_item", + "SELECT * FROM game_quest_reward_choice", + "SELECT * FROM game_creature_quest", + "SELECT * FROM game_gameobject_quest", + "SELECT * FROM game_trainer_spell", + "SELECT * FROM game_spell_chain", + "SELECT * FROM game_gameobject_template", + "SELECT * FROM game_talent", + "SELECT * FROM game_faction", +]; + +fn coordinator_queries(subscription: CoordinatorSubscription) -> Vec<&'static str> { let mut queries = crate::stdb::relays::queries(); queries.extend([ "SELECT * FROM game_realm", @@ -1017,53 +1071,26 @@ fn coordinator_queries(sharded_tables: bool) -> Vec<&'static str> { "SELECT * FROM game_meeting_stone", "SELECT * FROM game_meeting_stone_seeker", "SELECT * FROM game_config", - "SELECT * FROM game_creature_template", - "SELECT * FROM game_creature_spawn_map", - "SELECT * FROM game_area", - "SELECT * FROM game_taxi_node", - "SELECT * FROM game_taxi_path", - "SELECT * FROM game_taxi_path_node", "SELECT * FROM game_taxi_service_reply", - "SELECT * FROM game_start_position", "SELECT * FROM game_item_template", "SELECT * FROM game_item_random_property", "SELECT * FROM game_item_enchantment", - "SELECT * FROM game_spell", - "SELECT * FROM game_spell_effect", "SELECT * FROM game_character_buyback", "SELECT * FROM game_corpse_loot", - "SELECT * FROM game_npc_vendor", - "SELECT * FROM game_gossip_menu", - "SELECT * FROM game_gossip_condition", - "SELECT * FROM game_gossip_option", - "SELECT * FROM game_gossip_menu_profile", - "SELECT * FROM game_gossip_menu_profile_option", - "SELECT * FROM game_creature_gossip_menu_override", - "SELECT * FROM game_npc_text", - "SELECT * FROM game_npc_text_slot", - "SELECT * FROM game_quest_template", - "SELECT * FROM game_quest_text", - "SELECT * FROM game_quest_objective", - "SELECT * FROM game_quest_reward_item", - "SELECT * FROM game_quest_reward_choice", - "SELECT * FROM game_creature_quest", - "SELECT * FROM game_gameobject_quest", "SELECT * FROM game_player_spell", "SELECT * FROM game_player_action", - "SELECT * FROM game_trainer_spell", - "SELECT * FROM game_spell_chain", - "SELECT * FROM game_gameobject_template", - "SELECT * FROM game_talent", "SELECT * FROM game_character_talent", "SELECT * FROM game_mail", "SELECT * FROM game_mail_escrow", "SELECT * FROM game_item_text", - "SELECT * FROM game_faction", "SELECT * FROM game_faction_template", "SELECT * FROM game_guid_allocator", "SELECT * FROM game_guid_range", ]); - if sharded_tables { + if subscription != CoordinatorSubscription::RealmCore { + queries.extend_from_slice(WORLD_CATALOG_QUERIES); + } + if subscription != CoordinatorSubscription::SingleShard { queries.extend([ "SELECT * FROM game_character_shard", "SELECT * FROM game_group", @@ -1082,10 +1109,10 @@ fn connect_blocking( uri: String, db_name: String, token: Option, - sharded_tables: bool, + subscription: CoordinatorSubscription, park_for_relays: bool, ) -> Result { - let queries = coordinator_queries(sharded_tables); + let queries = coordinator_queries(subscription); let role_label = coordinator_role_label(&db_name); connect_subscribed( uri, @@ -1145,7 +1172,7 @@ fn spawn_coordinator_watchdog(inner: Arc) -> std::thread::Join inner.uri.clone(), inner.db_name.clone(), inner.token.clone(), - inner.sharded_tables, + inner.subscription, true, ) { Ok(fresh) => { @@ -1412,31 +1439,144 @@ pub(crate) fn recv_reducer_on( #[cfg(test)] mod coordinator_query_tests { - use super::{call_pipe_queries, coordinator_queries, CALL_PIPE_LIVENESS_QUERY}; + use super::{ + call_pipe_queries, coordinator_queries, CoordinatorSubscription, ShardMap, + CALL_PIPE_LIVENESS_QUERY, WORLD_CATALOG_QUERIES, + }; #[test] fn coordinator_subscribes_source_map_coverage_in_each_topology() { - for sharded in [false, true] { - assert!(coordinator_queries(sharded).contains(&"SELECT * FROM game_creature_spawn_map")); + for subscription in [ + CoordinatorSubscription::SingleShard, + CoordinatorSubscription::WorldShard, + ] { + assert!(coordinator_queries(subscription) + .contains(&"SELECT * FROM game_creature_spawn_map")); } } #[test] fn coordinator_subscribes_bot_intents_in_each_topology() { - for sharded in [false, true] { - let queries = coordinator_queries(sharded); + for subscription in [ + CoordinatorSubscription::SingleShard, + CoordinatorSubscription::WorldShard, + ] { + let queries = coordinator_queries(subscription); for query in [ "SELECT * FROM game_bot_invite_intent", "SELECT * FROM game_bot_transfer_intent", ] { assert!( queries.contains(&query), - "missing {query}, sharded={sharded}" + "missing {query}, subscription={subscription:?}" ); } } } + #[test] + fn realm_core_skips_world_catalogs_and_keeps_every_control_and_relay_query() { + let world = coordinator_queries(CoordinatorSubscription::WorldShard); + let realm = coordinator_queries(CoordinatorSubscription::RealmCore); + let removed: Vec<_> = world + .iter() + .filter(|query| !realm.contains(query)) + .copied() + .collect(); + assert_eq!(removed, WORLD_CATALOG_QUERIES); + for query in crate::stdb::relays::queries().into_iter().chain([ + "SELECT * FROM game_bot_invite_intent", + "SELECT * FROM game_bot_transfer_intent", + "SELECT * FROM game_party_command_intent WHERE pending = true", + "SELECT * FROM game_party_command_dispatch_lane", + "SELECT * FROM game_transfer_out", + "SELECT * FROM game_transfer_in", + "SELECT * FROM game_character_shard", + "SELECT * FROM game_account", + "SELECT * FROM game_account_fence", + "SELECT * FROM game_account_character_owner", + "SELECT * FROM game_item_template", + "SELECT * FROM game_item_random_property", + "SELECT * FROM game_item_enchantment", + "SELECT * FROM game_faction_template", + ]) { + assert!(realm.contains(&query), "missing Realm-core query {query}"); + } + } + + #[test] + fn a_realm_core_named_as_a_world_owner_keeps_world_catalogs() { + let map = ShardMap::parse("world", "0:*=realm").with_realm_core(Some("realm")); + let role = CoordinatorSubscription::for_shard(&map, "realm"); + assert_eq!(role, CoordinatorSubscription::WorldShard); + assert!(coordinator_queries(role).contains(&"SELECT * FROM game_spell")); + } + + #[test] + fn subscription_roles_keep_all_world_shards_ready_without_viewers() { + let map = ShardMap::parse("world", "0:*=world;1:*=idle").with_realm_core(Some("realm")); + assert_eq!(map.shards(), ["world", "idle"]); + for name in map.shards() { + let role = CoordinatorSubscription::for_shard(&map, &name); + assert_eq!(role, CoordinatorSubscription::WorldShard); + let queries = coordinator_queries(role); + assert!(queries.contains(&"SELECT * FROM game_spell")); + assert!(queries.contains(&"SELECT * FROM game_bot_invite_intent")); + assert!(queries.contains(&"SELECT * FROM game_bot_transfer_intent")); + } + assert_eq!( + CoordinatorSubscription::for_shard(&map, "realm"), + CoordinatorSubscription::RealmCore + ); + let single = ShardMap::parse("world", ""); + assert_eq!( + CoordinatorSubscription::for_shard(&single, "world"), + CoordinatorSubscription::SingleShard + ); + } + + #[test] + fn coordinator_reconnect_restores_the_saved_subscription_scope() { + fn restores_scope(source: &str) -> bool { + use syn::visit::Visit; + struct SavedScope(bool); + impl<'ast> Visit<'ast> for SavedScope { + fn visit_expr_call(&mut self, call: &'ast syn::ExprCall) { + if let syn::Expr::Path(path) = call.func.as_ref() { + if path.path.is_ident("connect_blocking") { + self.0 = matches!(call.args.iter().nth(3), Some(syn::Expr::Field(field)) + if matches!(&field.member, syn::Member::Named(member) if member == "subscription") + && matches!(field.base.as_ref(), syn::Expr::Path(base) if base.path.is_ident("inner"))); + } + } + syn::visit::visit_expr_call(self, call); + } + } + let file = syn::parse_file(source).unwrap(); + let watchdog = file + .items + .into_iter() + .find_map(|item| match item { + syn::Item::Fn(function) + if function.sig.ident == "spawn_coordinator_watchdog" => + { + Some(function) + } + _ => None, + }) + .unwrap(); + let mut scope = SavedScope(false); + scope.visit_block(&watchdog.block); + scope.0 + } + let source = include_str!("connection.rs"); + assert!(restores_scope(source)); + assert!(!restores_scope(&source.replace( + "inner.subscription,", + "CoordinatorSubscription::WorldShard,", + ))); + } + const MULTI_DB_TABLES: &[&str] = &[ "SELECT * FROM game_character_shard", "SELECT * FROM game_group", @@ -1449,7 +1589,7 @@ mod coordinator_query_tests { #[test] fn a_single_database_gateway_subscribes_no_multi_database_table() { - let single = coordinator_queries(false); + let single = coordinator_queries(CoordinatorSubscription::SingleShard); for table in MULTI_DB_TABLES { assert!( !single.contains(table), @@ -1461,7 +1601,7 @@ mod coordinator_query_tests { #[test] fn loot_tag_projection_rows_are_in_every_base_subscription() { - let queries = coordinator_queries(false); + let queries = coordinator_queries(CoordinatorSubscription::SingleShard); for table in [ "SELECT * FROM game_creature_quest_tap", "SELECT * FROM game_creature_quest_tap_member", @@ -1478,8 +1618,8 @@ mod coordinator_query_tests { #[test] fn the_multi_database_flag_adds_exactly_those_tables_and_changes_nothing_else() { - let single = coordinator_queries(false); - let sharded = coordinator_queries(true); + let single = coordinator_queries(CoordinatorSubscription::SingleShard); + let sharded = coordinator_queries(CoordinatorSubscription::WorldShard); assert_eq!( sharded.len(), single.len() + MULTI_DB_TABLES.len(), @@ -1512,7 +1652,7 @@ mod coordinator_query_tests { "a reducer-only pipe must carry only its game_config liveness signal" ); - let coordinator_queries = coordinator_queries(false); + let coordinator_queries = coordinator_queries(CoordinatorSubscription::SingleShard); assert!(coordinator_queries.len() > pipe_queries.len()); assert!(coordinator_queries.contains(&CALL_PIPE_LIVENESS_QUERY)); } @@ -2150,7 +2290,7 @@ impl Coordinator { async fn connect_shard( cfg: &GatewayConfig, db_name: &str, - sharded_tables: bool, + subscription: CoordinatorSubscription, ) -> Result> { let uri = cfg.stdb_uri.clone(); let db = db_name.to_string(); @@ -2159,7 +2299,7 @@ impl Coordinator { let join_role_label = role_label.clone(); let build_thread = std::thread::Builder::new() .name("stdb-coordinator-connect".into()) - .spawn(move || connect_blocking(uri, db, token, sharded_tables, false)) + .spawn(move || connect_blocking(uri, db, token, subscription, false)) .context(format!("spawn {role_label} connect thread"))?; let live = tokio::task::spawn_blocking(move || -> Result { build_thread @@ -2208,7 +2348,7 @@ impl Coordinator { token: cfg.coordinator_token.clone(), uri: cfg.stdb_uri.clone(), db_name: db_name.to_string(), - sharded_tables, + subscription, call_pipes, call_pipe_next: std::sync::atomic::AtomicUsize::new(0), party_command_lane_overflow: AtomicBool::new(false), @@ -2264,10 +2404,9 @@ impl Coordinator { ); } let map = ShardMap::from_env(&cfg.module_name); - let sharded_tables = map.databases().len() > 1; let mut conns: HashMap> = HashMap::new(); for db in map.databases() { - match Self::connect_shard(cfg, &db, sharded_tables).await { + match Self::connect_shard(cfg, &db, CoordinatorSubscription::for_shard(&map, &db)).await { Ok(inner) => { log::info!("coordinator connected to shard {db}"); conns.insert(db, inner); diff --git a/gateway/src/stdb/mod.rs b/gateway/src/stdb/mod.rs index 4845046c..256b4ca2 100644 --- a/gateway/src/stdb/mod.rs +++ b/gateway/src/stdb/mod.rs @@ -24,3 +24,6 @@ pub use subscriptions::PlayerSubscriptions; pub use views::{AccountRow, RealmRow}; pub(crate) mod relays; + +#[cfg(test)] +mod subscription_scope_durable_tests; diff --git a/gateway/src/stdb/subscription_scope_durable_tests.rs b/gateway/src/stdb/subscription_scope_durable_tests.rs new file mode 100644 index 00000000..c5d775c0 --- /dev/null +++ b/gateway/src/stdb/subscription_scope_durable_tests.rs @@ -0,0 +1,272 @@ +use std::sync::{mpsc, Arc}; + +use spacetimedb_sdk::{DbContext, Table}; + +use crate::accept::BlockingTaskCapacity; +use crate::config::GatewayConfig; +use crate::durable_test_support::{module_bytes, Standalone, POLL_TIMEOUT}; +use crate::stdb::bindings::*; +use crate::stdb::Coordinator; +use crate::world::{ + Actor, AuctionActionStore, AuctionBrowseRequest, AuctionQuery, CastStore, MailStore, +}; + +const REALM: &str = "subscription-scope-realm"; +const TOPOLOGY_VARS: [&str; 3] = [ + "LYRACORE_SHARD_MAP", + "LYRACORE_SHARD_MAP_FILE", + "LYRACORE_REALM_CORE", +]; + +struct TopologyEnv { + previous: [Option; 3], + _guard: std::sync::MutexGuard<'static, ()>, +} + +impl TopologyEnv { + fn install(mixed_role: bool) -> Self { + let guard = super::subscriptions::DURABLE_TOPOLOGY_ENV_LOCK + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let previous = TOPOLOGY_VARS.map(std::env::var_os); + std::env::remove_var("LYRACORE_SHARD_MAP_FILE"); + std::env::set_var( + "LYRACORE_SHARD_MAP", + if mixed_role { + format!("1:*={REALM}") + } else { + String::new() + }, + ); + std::env::set_var("LYRACORE_REALM_CORE", REALM); + Self { + previous, + _guard: guard, + } + } +} + +impl Drop for TopologyEnv { + fn drop(&mut self) { + for (name, previous) in TOPOLOGY_VARS.into_iter().zip(&self.previous) { + match previous { + Some(value) => std::env::set_var(name, value), + None => std::env::remove_var(name), + } + } + } +} + +fn catalogue_counts(coordinator: &Coordinator) -> Vec<(&'static str, usize)> { + let live = coordinator.0.coord(); + let db = &live.conn.db; + vec![ + ( + "game_creature_template", + db.game_creature_template().iter().count(), + ), + ( + "game_start_position", + db.game_start_position().iter().count(), + ), + ("game_spell", db.game_spell().iter().count()), + ("game_spell_effect", db.game_spell_effect().iter().count()), + ("game_npc_vendor", db.game_npc_vendor().iter().count()), + ( + "game_quest_template", + db.game_quest_template().iter().count(), + ), + ( + "game_quest_objective", + db.game_quest_objective().iter().count(), + ), + ( + "game_gameobject_template", + db.game_gameobject_template().iter().count(), + ), + ("game_taxi_node", db.game_taxi_node().iter().count()), + ] +} + +fn assert_scope(shard: &Standalone, realm: &Coordinator, mixed_role: bool) { + for (table, cached) in catalogue_counts(realm) { + let durable = shard + .query_database_rows(REALM, &format!("SELECT * FROM {table}")) + .len(); + assert!(durable > 0, "the Module fixture must populate {table}"); + assert_eq!(cached, if mixed_role { durable } else { 0 }, "{table}"); + } + let live = realm.0.coord(); + assert_eq!( + live.conn + .db + .game_faction_template() + .id() + .find(&5090104) + .unwrap() + .faction, + 7 + ); + assert_eq!( + live.conn + .db + .game_item_random_property() + .property_id() + .find(&5090101) + .unwrap() + .enchant_id_1, + 5090103 + ); + assert_eq!( + live.conn + .db + .game_item_enchantment() + .id() + .find(&1303066368) + .unwrap() + .amount, + 7 + ); +} + +fn assert_retained_store_reads( + world: &Coordinator, + realm: &Coordinator, + max_durability: u32, + bid: u32, +) { + let item = realm.item_template(5090050).unwrap().unwrap(); + assert_eq!(item.max_durability, max_durability); + assert!(item.name.starts_with("Scope Blade")); + let mut mail = realm.mail_list(1).unwrap(); + mail.sort_by_key(|letter| letter.subject.clone()); + assert_eq!(mail.len(), 3); + let attached: Vec<_> = mail + .iter() + .filter(|letter| letter.item_entry != 0) + .collect(); + assert_eq!(attached.len(), 2); + for letter in attached { + assert_eq!(letter.item_entry, 5090050); + assert_eq!(letter.item_stack_count, 1); + assert_eq!(letter.item_durability, 17); + assert_eq!(letter.max_durability, max_durability); + assert_eq!(realm.mail_by_id(letter.id).unwrap().unwrap(), *letter); + } + let page = world + .auction_query( + Actor::new(1).unwrap(), + 1, + AuctionQuery::Browse(AuctionBrowseRequest { + auctioneer_guid: 0, + offset: 0, + name: "Scope Blade".into(), + minimum_level: None, + maximum_level: None, + inventory_type: None, + item_class: None, + item_subclass: None, + quality: None, + usable_only: false, + }), + ) + .unwrap(); + assert_eq!(page.total, 1); + let [auction] = page.rows.as_slice() else { + panic!("expected one retained auction, got {:?}", page.rows); + }; + assert_eq!(auction.id, 5090050); + assert_eq!(auction.owner_guid, 5090050); + assert_eq!(auction.item_entry, 5090050); + assert_eq!(auction.item_stack_count, 2); + assert_eq!(auction.random_property_id, 117); + assert_eq!(auction.highest_bid, bid); +} + +fn verify_subscription_scope(mixed_role: bool) { + let _topology = TopologyEnv::install(mixed_role); + let mut shard = Standalone::start_persistent("subscription-scope"); + shard.publish_module(); + shard.publish_named_module_bytes(REALM, module_bytes()); + shard.assert_call("claim_operator", &[]); + shard.assert_call_database(REALM, "claim_operator", &[]); + shard.assert_call("debug_seed_scenario_fixtures", &[]); + shard.assert_call_database(REALM, "debug_seed_scenario_fixtures", &[]); + shard.assert_call("debug_spawn_player_entity", &["1"]); + shard.assert_call_database(REALM, "debug_stage_mail_take_fixture", &[]); + shard.assert_call_database(REALM, "debug_stage_auction_buyout_fixture", &[]); + shard.assert_sql("UPDATE game_item_template SET name = 'World Blade' WHERE entry = 5090050"); + shard.assert_sql_database(REALM, "UPDATE game_item_template SET name = 'Scope Blade', max_durability = 77 WHERE entry = 5090050"); + shard.assert_sql_database(REALM, "INSERT INTO game_faction_template (id,faction,faction_group,friend_group,enemy_group,enemy_0,enemy_1,enemy_2,enemy_3,friend_0,friend_1,friend_2,friend_3) VALUES (5090104,7,0,0,0,0,0,0,0,0,0,0,0)"); + shard.assert_sql_database(REALM, "INSERT INTO game_item_random_property (property_id,enchant_id_1,enchant_id_2,enchant_id_3,suffix) VALUES (5090101,5090103,0,0,'of the Fixture')"); + shard.assert_sql_database(REALM, "INSERT INTO game_item_enchantment (id,enchant_id,effect_index,kind,amount,spell_id,school_mask) VALUES (1303066368,5090103,0,14,7,0,0)"); + let cfg = GatewayConfig { + logon_bind: "127.0.0.1:0".into(), + world_bind: "127.0.0.1:0".into(), + stdb_uri: shard.server().into(), + module_name: shard.shard_name().into(), + coordinator_token: Some(shard.owner_token()), + gateway_id: "subscription-scope-test".into(), + blocking_task_capacity: BlockingTaskCapacity::new(1), + }; + let runtime = tokio::runtime::Runtime::new().unwrap(); + let world = runtime.block_on(Coordinator::connect(&cfg)).unwrap(); + let realm = world.realm_core().unwrap(); + assert_eq!(realm.shard_name(), REALM); + assert_eq!( + world.shard_for(1, 0).as_ref().map(Coordinator::shard_name), + mixed_role.then_some(REALM) + ); + assert!(catalogue_counts(&world).iter().all(|(_, count)| *count > 0)); + assert_scope(&shard, &realm, mixed_role); + assert_retained_store_reads(&world, &realm, 77, 201); + let generation = realm + .0 + .coord() + .durable_character_presence_snapshot(1, REALM) + .unwrap() + .generation; + + realm.park_pump_for_test(); + shard.assert_sql_database(REALM, "UPDATE game_item_template SET name = 'Scope Blade Rebuilt', max_durability = 88 WHERE entry = 5090050"); + shard.assert_sql_database( + REALM, + "UPDATE game_auction SET highest_bid = 251, revision = 4 WHERE id = 5090050", + ); + shard.assert_sql_database( + REALM, + "UPDATE game_spell SET name = 'Scope Rebuilt' WHERE spell_id = 50310", + ); + let (reconnected, ready) = mpsc::channel(); + realm.0.on_reconnect.lock().unwrap().push(Arc::new(move || { + let _ = reconnected.send(()); + })); + realm.0.coord().conn.disconnect().unwrap(); + realm.resume_pump_for_test(); + ready.recv_timeout(POLL_TIMEOUT).unwrap(); + let fresh = realm + .0 + .coord() + .durable_character_presence_snapshot(1, REALM) + .unwrap(); + assert!(fresh.generation > generation); + assert_scope(&shard, &realm, mixed_role); + assert_retained_store_reads(&world, &realm, 88, 251); + let spell = realm.0.coord().conn.db.game_spell().spell_id().find(&50310); + assert_eq!( + spell.map(|row| row.name), + mixed_role.then(|| "Scope Rebuilt".into()) + ); +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn dedicated_realm_core_keeps_retained_store_reads_and_omits_world_catalogues_after_reconnect() { + verify_subscription_scope(false); +} + +#[test] +#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"] +fn realm_core_that_also_owns_a_world_map_keeps_world_catalogues_after_reconnect() { + verify_subscription_scope(true); +} diff --git a/gateway/tests/schema_parity.rs b/gateway/tests/schema_parity.rs index 43d6ad9f..ae331c49 100644 --- a/gateway/tests/schema_parity.rs +++ b/gateway/tests/schema_parity.rs @@ -1153,10 +1153,14 @@ fn extract_select_star_tables(source: &str) -> Vec { let file = syn::parse_file(source).expect("valid query source"); let mut queries = Queries(Vec::new()); for item in file.items { - if let syn::Item::Fn(function) = item { - if function.sig.ident == "coordinator_queries" { + match item { + syn::Item::Fn(function) if function.sig.ident == "coordinator_queries" => { queries.visit_block(&function.block); } + syn::Item::Const(catalogs) if catalogs.ident == "WORLD_CATALOG_QUERIES" => { + queries.visit_expr(&catalogs.expr); + } + _ => {} } } queries.0 @@ -1227,12 +1231,13 @@ fn completeness_inventory_detects_an_unmanifested_read() { fn coordinator_queries() { ["SELECT * FROM game_real_one", "SELECT * FROM game_missing_read WHERE pending = true"]; } + const WORLD_CATALOG_QUERIES: &[&str] = &["SELECT * FROM game_missing_catalog"]; #[cfg(test)] fn unrelated_test() { "SELECT * FROM game_test_only"; } "#; assert_eq!( missing_manifest_tables(&extract_select_star_tables(source), &["game_real_one"]), - ["game_missing_read"], + ["game_missing_read", "game_missing_catalog"], ); }