fix(loot): retry failed settlements while results remain

gateway/src/world/loot.rs

@@ -208,15 +208,15 @@ pub(crate) fn flush_pending_promotions<St: WorldStore + ?Sized>(
 /// One promotion + settlement pass. No-op on an unsharded store (`realm_store()` answers
 /// `None`), which is what makes running this on a timer free for a single-database gateway.
 ///
-/// `won_watermark` is the caller's own `game_group_event.id` high-water mark, threaded through
-/// call to call so the relay never re-settles an old win after a restart within one process
-/// lifetime forgets nothing it already saw — advanced only up to what [`WorldStore::loot_won_since`]
-/// actually returns, never optimistically.
+/// `won_watermark` advances only after every settlement in the batch succeeds on every Shard.
+/// A failure leaves the result available to retry while its Realm-core event still exists.
+/// A successful grant removes the original loot row, and the Module ignores rows that are not
+/// withheld. These guards cannot distinguish a later loot row with the same corpse GUID and slot.
 ///
 /// Both directions are BEST-EFFORT, deliberately, the same posture `party::sync_mirrors` documents:
 /// realm-core has already committed (a promoted roll exists there, or a roll has already resolved)
 /// by the time either loop runs, so a failed relay step must not undo or re-litigate that — it just
-/// leaves the affected shard's local state stale until the next tick.
+/// leaves the affected Shard's local state stale. Settlement retries still need the outcome event.
 ///
 /// Ordinary promotion latency (a roll NOT caught by [`flush_pending_promotions`]) is bounded by this
 /// function's own caller's poll interval, not by anything in here — see this module's doc for why
@@ -232,18 +232,22 @@ pub(crate) fn relay_tick<St: WorldStore + ?Sized>(store: &St, won_watermark: &mu
     // harmless no-ops, so this does not need to know in advance which one holds the corpse.
     match realm.loot_won_since(*won_watermark) {
         Ok((new_watermark, wins)) => {
+            let mut settled = true;
             for (corpse_guid, slot, winner_guid) in wins {
                 for shard in store.world_stores() {
                     if let Err(e) = shard.settle_loot_roll(corpse_guid, slot, winner_guid) {
+                        settled = false;
                         log::warn!(
-                            "loot-roll relay: could not settle ({corpse_guid}, {slot}) on {} ({e:#}) \
-                             — the winner's item stays on the corpse until the next successful settle",
+                            "loot-roll relay: could not settle ({corpse_guid}, {slot}) on {} ({e:#}); \
+                             retrying while the result event remains",
                             shard.shard_name()
                         );
                     }
                 }
             }
-            *won_watermark = new_watermark;
+            if settled {
+                *won_watermark = new_watermark;
+            }
         }
         Err(e) => {
             log::warn!("loot-roll relay: could not read realm-core's ROLL_WON events ({e:#})")

gateway/src/world/loot_tests.rs

@@ -231,6 +231,34 @@ fn relay_tick_settles_every_connected_shard_for_a_resolved_roll() {
     );
 }
 
+#[test]
+fn a_failed_settlement_retries_after_the_shard_recovers() {
+    let (realm, world, instances, _) = party_topology();
+    let unavailable = std::sync::Arc::new(InMemoryStore {
+        shard: instances.shard.clone(),
+        settle_loot_roll_error: Some("Shard connection unavailable".into()),
+        ..Default::default()
+    });
+    *world.peers.lock().unwrap() = vec![unavailable.clone(), world.clone()];
+    let win = (500, 2, GINGER);
+    *realm.won_events.lock().unwrap() = vec![win];
+    let mut watermark = 0;
+
+    loot::relay_tick(world.as_ref(), &mut watermark);
+
+    assert!(unavailable.settled_rolls.lock().unwrap().is_empty());
+    assert_eq!(*world.settled_rolls.lock().unwrap(), vec![win]);
+    assert_eq!(watermark, 0, "the failed Shard still needs this result");
+
+    *world.peers.lock().unwrap() = vec![instances.clone(), world.clone()];
+    loot::relay_tick(world.as_ref(), &mut watermark);
+
+    assert_eq!(*instances.settled_rolls.lock().unwrap(), vec![win]);
+    assert_eq!(watermark, 1);
+    loot::relay_tick(world.as_ref(), &mut watermark);
+    assert_eq!(*instances.settled_rolls.lock().unwrap(), vec![win]);
+}
+
 // ---- `flush_pending_promotions` (the disband race, found in adversarial review) ----
 
 /// **AC: a disband-capable op promotes every connected shard's pending rolls BEFORE it reaches