perf(module): stop retaining transient notification rows

.github/workflows/module-durable.yml

@@ -68,6 +68,7 @@ jobs:
           cargo test -p lyracore-gateway --bin lyracore-gateway stdb::armor::tests::resistance_projection_adds_property_effects_only_while_the_item_is_worn_and_unbroken -- --ignored --exact
           cargo test -p lyracore-gateway --test provision_cli provision_writes_the_account_to_the_world_shard_and_realm_core -- --ignored --exact
           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::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

.github/workflows/rust.yml

@@ -53,6 +53,7 @@ jobs:
       - run: cargo test -p lyracore-importer
       - run: cargo test -p lyracore-module --lib --features=debug_reducers
       - run: cargo test -p lyracore-module --test build_script
+      - run: cargo test -p lyracore-module --test notification_tables
       - run: cargo test -p lyracore-module --test package_api_lint
       - run: cargo test -p lyracore-package-delta
       - run: cargo test -p lyracore-gateway

crates/lyracore-test-support/src/lib.rs

@@ -1,7 +1,9 @@
 mod module_wasm;
 pub mod source_scan;
+mod subscription;
 
 pub use module_wasm::{gateway_binary, module_bytes};
+pub use subscription::TableSubscription;
 
 use std::collections::{BTreeMap, BTreeSet};
 use std::ffi::OsString;
@@ -268,6 +270,10 @@ impl Standalone {
         command.output().expect("failed to call reducer")
     }
 
+    pub fn subscribe(&self, queries: &[&str]) -> TableSubscription {
+        TableSubscription::start(self.command(), &self.server, &self.database, queries)
+    }
+
     pub fn capture_updates(
         &self,
         query: &str,

crates/lyracore-test-support/src/subscription.rs

@@ -0,0 +1,87 @@
+use std::io::{BufRead, BufReader};
+use std::process::{Child, Command, Stdio};
+use std::sync::{Arc, Mutex};
+use std::thread::{self, JoinHandle};
+
+use serde_json::Value;
+
+pub struct TableSubscription {
+    child: Child,
+    reader: Option<JoinHandle<()>>,
+    updates: Arc<Mutex<Result<Vec<Value>, String>>>,
+}
+
+impl TableSubscription {
+    pub(crate) fn start(mut command: Command, server: &str, shard: &str, queries: &[&str]) -> Self {
+        let child = command
+            .args([
+                "subscribe",
+                "-s",
+                server,
+                "--print-initial-update",
+                "--timeout",
+                "180",
+                shard,
+            ])
+            .args(queries)
+            .stdout(Stdio::piped())
+            .spawn()
+            .expect("failed to subscribe to fixture");
+        let mut subscription = Self {
+            child,
+            reader: None,
+            updates: Arc::new(Mutex::new(Ok(Vec::new()))),
+        };
+        let mut reader = BufReader::new(subscription.child.stdout.take().unwrap());
+        let mut initial = String::new();
+        assert!(reader.read_line(&mut initial).unwrap() > 0);
+        let _: Value = serde_json::from_str(&initial).expect("initial subscription result");
+        let recorded = subscription.updates.clone();
+        subscription.reader = Some(thread::spawn(move || {
+            for line in reader.lines() {
+                let update = line.map_err(|error| error.to_string()).and_then(|line| {
+                    serde_json::from_str(&line).map_err(|error| error.to_string())
+                });
+                match update {
+                    Ok(update) => recorded.lock().unwrap().as_mut().unwrap().push(update),
+                    Err(error) => {
+                        *recorded.lock().unwrap() = Err(error);
+                        return;
+                    }
+                }
+            }
+            *recorded.lock().unwrap() = Err("fixture subscription ended".into());
+        }));
+        subscription
+    }
+
+    pub fn inserts(&self, table: &str) -> Vec<Value> {
+        self.rows(table, "inserts")
+    }
+
+    pub fn deletes(&self, table: &str) -> Vec<Value> {
+        self.rows(table, "deletes")
+    }
+
+    fn rows(&self, table: &str, change: &str) -> Vec<Value> {
+        self.updates
+            .lock()
+            .unwrap()
+            .as_ref()
+            .expect("fixture subscription failed")
+            .iter()
+            .flat_map(|update| update[table][change].as_array().into_iter().flatten())
+            .cloned()
+            .collect()
+    }
+}
+
+impl Drop for TableSubscription {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+        if let Some(reader) = self.reader.take() {
+            let _ = reader.join();
+        }
+    }
+}

docs/architecture.md

@@ -584,6 +584,12 @@ table's insert, update and delete callbacks and registered before the subscripti
 per Shard). Each lives on its connection generation, so a reconnect rebuilds it from the fresh
 subscription.
 
+Combat, spell cast, spell impact, Speech and emote notifications use native `*_event_live` tables.
+The ordinary predecessors retain their published schemas for additive upgrades and expire through the
+event reaper. The Gateway handles inserts without caching or replaying notification rows. Group events
+still use their ordered committed batch, and login still reads retained System Messages. Read
+[danger-zones.md](./danger-zones.md) before publishing the matching Module and Gateway.
+
 ### 5.4 RLS is not a universal backstop — state this plainly
 
 - **Anyone who can reach the node's `:3000` port** can connect with no token, get a minted

docs/danger-zones.md

@@ -84,6 +84,38 @@ any line in §1 needs a human review before it ships, whoever or whatever wrote
    realm-wide or partition-managing read, with a one-line justification, and delete budget as scans
    are replaced with helpers — a budget larger than the real count fails its own test.
 
+### Native Relay events
+
+Combat, spell cast, spell impact, Speech and emote notifications use native SpacetimeDB event
+tables. Each adds `_live` to its original table name, such as `game_combat_event_live`.
+The original `game_combat_event`, `game_spell_cast_event`, `game_spell_impact_event`,
+`game_chat_event` and `game_emote_event` remain ordinary tables with their published schemas.
+The two tables share each row type. The Module writes new notifications only to the native table.
+The event reaper expires rows left in the original table after an upgrade.
+
+SpacetimeDB refuses a change to an existing table's event flag. Keep the old declarations and
+their indexes when publishing this additive change. Stop the Gateway, publish the Module to
+every configured Shard, then start the matching Gateway. An old Gateway subscribes to the old
+names and will miss new notifications. Do not clear a Shard to make this migration apply.
+
+Native events reach connected subscribers through insert callbacks. They have no initial replay,
+cached rows, update callbacks or delete callbacks. The Gateway uses the same row fields, audience
+rules and packet rendering for the new tables. Cast notifications carry their start, completion
+or outcome in each row; the Relays do not read a retained event batch. A reconnect restores
+durable state and receives subsequent notifications. Events emitted during a Transport Loss
+are not replayed. Event inserts still enter the commit log, so this change does not replace
+history retention. See the [event table contract](https://spacetimedb.com/docs/tables/event-tables/).
+
+Keep `game_group_event` ordinary. `GroupEventCursor` reads the committed batch and sorts its ids
+before delivery. Keep `game_system_message_event` ordinary because login reads pending messages.
+Pending casts and impacts, Transfer Intent, party command intent and receipts, Mail, invitations
+and other durable work keep their current storage and completion rules.
+
+Durable tests observe notifications through subscriptions opened before the action. A SQL read
+after the action sees an empty native event table. The old debug reducer
+`debug_verify_ranged_lethal_damage_floor_fixture` was removed because it read retained cast logs.
+Its durable test now checks the subscribed damage event and the target's durable health.
+
 ## 2. Tooling gotchas (will silently mislead you)
 
 - **`spacetime sql` (2.5):** NO `ORDER BY`, NO `IN`/subqueries, NO `Timestamp`/`Identity` literal. A

docs/schema.md

@@ -225,9 +225,9 @@ scan on a sharded realm silently returns a subset rather than erroring, and ever
 `module/src/tripwires.rs::partition_discipline_tripwire` against a whitelist that only ever shrinks.
 
 Six sibling tables carry the **identical** `(map_id, instance_id, grid_x, grid_y)` key so they can
-ride the same AOI box: `game_entity_motion`, `game_creature_spline`, `game_combat_event`,
-`game_spell_cast_event`, `game_spell_impact_event`, `game_emote_event`. `game_roll_event` keeps the
-same key, but nothing writes it: `/roll` is a Group Broadcast.
+ride the same AOI box: `game_entity_motion`, `game_creature_spline`, `game_combat_event_live`,
+`game_spell_cast_event_live`, `game_spell_impact_event_live`, `game_emote_event_live`.
+`game_roll_event` keeps the same key, but nothing writes it: `/roll` is a Group Broadcast.
 `game_gameobject`'s grid key is only three columns (no `instance_id`).
 
 ### `game_character` — the durable character (`module/src/character.rs:7`)
@@ -307,7 +307,7 @@ whose ownership has closed or expired can use null ownership.
 
 ### Chat (`module/src/chat.rs`, `module/src/realm_chat.rs`, `module/src/channel.rs`)
 
-`game_chat_event` is the public `[event]` row for one say, yell, `/e` or creature line on the
+`game_chat_event_live` is the public native event row for one say, yell, `/e` or creature line on the
 speaker's own Shard. The Gateway scopes it to listeners in range when it relays it. Only
 `apply_send_chat` (a Character) and `apply_send_chat_to` (EventAI) write it. A Character's line
 passes the language Gate first.

docs/testing.md

@@ -30,6 +30,7 @@ cargo clippy --workspace --all-targets --all-features -- -D warnings
 cargo test -p lyracore-importer
 cargo test -p lyracore-module --lib --features=debug_reducers
 cargo test -p lyracore-module --test build_script
+cargo test -p lyracore-module --test notification_tables
 cargo test -p lyracore-module --test package_api_lint
 cargo test -p lyracore-package-delta
 cargo test -p lyracore-gateway
@@ -49,8 +50,8 @@ compile checks their types and generated registrations.
 
 ## Durable tier
 
-From `.github/workflows/module-durable.yml`. Each test in `module/tests/` is `#[ignore]`d because it
-starts its own SpacetimeDB standalone on a free loopback port, publishes the Module with
+From `.github/workflows/module-durable.yml`. Durable tests in `module/tests/` are `#[ignore]`d because
+each starts its own SpacetimeDB standalone on a free loopback port, publishes the Module with
 `--features=debug_reducers,package_test_fixture`, calls reducers and reads the result. The
 `lyracore-test-support` crate owns that fixture. Standalone logs go to `lyracore-standalone-logs`
 under the system temp directory.

gateway/src/codec/combat/subscriptions.rs

@@ -1139,18 +1139,18 @@ crate::stdb::relays::declare_relays! {
         insert(Snapshot) => |context, _event, row| dynobj_appeared(&context.view, context.shard, row);
         delete(Snapshot) => |context, _event, row| dynobj_vanished(&context.view, context.shard, row);
     }
-    game_combat_event [false] {
+    game_combat_event_live [false] {
         insert(Snapshot) => |context, _event, row| combat_event_appeared(&context.view, context.shard, row);
     }
     game_melee_attack [false] {
         insert(Snapshot) => |context, event, row| melee_engaged(&context.view, context.shard, row, !matches!(event.event, spacetimedb_sdk::Event::SubscribeApplied));
         update(Snapshot) => |context, _event, old, row| melee_updated(&context.view, context.shard, old, row);
         delete(Snapshot) => |context, _event, row| melee_disengaged(&context.view, context.shard, row);
     }
-    game_spell_cast_event [false] {
+    game_spell_cast_event_live [false] {
         insert(Snapshot) => |context, _event, row| cast_event_appeared(&context.view, context.shard, row);
     }
-    game_spell_impact_event [false] {
+    game_spell_impact_event_live [false] {
         insert(Snapshot) => |context, _event, row| impact_appeared(&context.view, context.shard, row);
     }
     game_aura [false] {

gateway/src/codec/social/subscriptions.rs

@@ -1102,10 +1102,10 @@ crate::stdb::relays::declare_relays! {
         insert(Snapshot) => |context, _event, row| group_member_mirror_changed(&context.view, &context.coord, context.shard, row);
         delete(Snapshot) => |context, _event, row| group_member_mirror_changed(&context.view, &context.coord, context.shard, row);
     }
-    game_emote_event [false] {
+    game_emote_event_live [false] {
         insert(Snapshot) => |context, _event, row| emote_appeared(&context.view, &context.coord, context.shard, row);
     }
-    game_chat_event [false] {
+    game_chat_event_live [false] {
         insert(Snapshot) => |context, _event, row| chat_appeared(&context.view, &context.coord, context.shard, row);
     }
     game_group_event [true] {

gateway/src/stdb/relays.rs

@@ -1,6 +1,7 @@
 use std::panic::{catch_unwind, AssertUnwindSafe};
 use std::sync::Arc;
 
+use spacetimedb_sdk::table::WithInsert;
 use spacetimedb_sdk::{Table, TableWithPrimaryKey};
 
 use super::bindings::{EventContext, Reducer, RemoteTables};
@@ -82,7 +83,7 @@ fn guarded(label: &str, body: impl FnOnce()) {
     }
 }
 
-pub(crate) fn insert<T: Table<EventContext = EventContext>>(
+pub(crate) fn insert<T: WithInsert<EventContext = EventContext>>(
     table: T,
     label: &'static str,
     context: RelayContext,

gateway/tests/notification_events.rs

@@ -0,0 +1,123 @@
+include!(concat!(env!("OUT_DIR"), "/bindings.rs"));
+
+use bindings::{DbConnection, GameChatEventLiveTableAccess, SubscriptionHandle};
+use lyracore_test_support::{actor, Standalone, POLL_TIMEOUT};
+use spacetimedb_sdk::{DbContext, EventTable};
+use std::sync::mpsc::{self, Receiver};
+use std::thread::JoinHandle;
+
+struct Delivery {
+    message: String,
+    cached_count: u64,
+    cached_rows: usize,
+}
+
+struct Subscribed {
+    connection: DbConnection,
+    pump: Option<JoinHandle<()>>,
+    _subscription: SubscriptionHandle,
+    deliveries: Receiver<Delivery>,
+}
+
+impl Subscribed {
+    fn connect(shard: &Standalone) -> Self {
+        let connection = DbConnection::builder()
+            .with_uri(shard.server())
+            .with_database_name(shard.shard_name())
+            .with_token(Some(shard.owner_token()))
+            .build()
+            .expect("fixture connection");
+        let (delivery, deliveries) = mpsc::channel();
+        connection
+            .db
+            .game_chat_event_live()
+            .on_insert(move |ctx, row| {
+                delivery
+                    .send(Delivery {
+                        message: row.message.clone(),
+                        cached_count: ctx.db.game_chat_event_live().count(),
+                        cached_rows: ctx.db.game_chat_event_live().iter().count(),
+                    })
+                    .expect("fixture receiver remains connected");
+            });
+        let (applied, ready) = mpsc::channel();
+        let failed = applied.clone();
+        let subscription = connection
+            .subscription_builder()
+            .on_applied(move |_| {
+                applied.send(Ok(())).unwrap();
+            })
+            .on_error(move |_, error| {
+                failed.send(Err(error.to_string())).unwrap();
+            })
+            .subscribe(["SELECT * FROM game_chat_event_live"]);
+        let pump = connection.run_threaded();
+        let subscribed = Self {
+            connection,
+            pump: Some(pump),
+            _subscription: subscription,
+            deliveries,
+        };
+        ready
+            .recv_timeout(POLL_TIMEOUT)
+            .expect("subscription must apply")
+            .expect("subscription must succeed");
+        subscribed
+    }
+
+    fn expect_line(&self, message: &str) {
+        let delivery = self
+            .deliveries
+            .recv_timeout(POLL_TIMEOUT)
+            .expect("Speech insert");
+        assert_eq!(delivery.message, message);
+        assert_eq!(delivery.cached_count, 0);
+        assert_eq!(delivery.cached_rows, 0);
+    }
+}
+
+impl Drop for Subscribed {
+    fn drop(&mut self) {
+        let _ = self.connection.disconnect();
+        if let Some(pump) = self.pump.take() {
+            let _ = pump.join();
+        }
+    }
+}
+
+fn speak(shard: &Standalone, message: &str) {
+    shard.assert_call(
+        "gw_send_chat",
+        &[
+            &actor("1"),
+            "0",
+            "7",
+            &serde_json::to_string(message).unwrap(),
+        ],
+    );
+}
+
+#[test]
+#[ignore = "requires the SpacetimeDB 2.7.1 CLI and Wasm toolchain"]
+fn native_notifications_keep_transaction_order_without_cached_rows_or_reconnect_replay() {
+    let mut shard = Standalone::start("notification-events");
+    shard.publish_module();
+    shard.assert_call("claim_operator", &[]);
+    shard.assert_call("debug_spawn_player_entity", &["1"]);
+    let first = Subscribed::connect(&shard);
+    speak(&shard, "first");
+    speak(&shard, "second");
+    first.expect_line("first");
+    first.expect_line("second");
+    assert!(shard
+        .query_rows("SELECT * FROM game_chat_event_live")
+        .is_empty());
+    assert!(shard.query_rows("SELECT * FROM game_chat_event").is_empty());
+    drop(first);
+
+    speak(&shard, "while disconnected");
+    let next = Subscribed::connect(&shard);
+    assert!(next.deliveries.try_recv().is_err());
+    speak(&shard, "after reconnect");
+    next.expect_line("after reconnect");
+}