cas: fix false GC-lease steal — heartbeat seq compared across different hb owners

The steal gate's liveness check compared the observed `gc/hb` sequence
against the previous observation while matching the hb OWNER against the
CURRENT LEASE owner:

    hb_alive = has_observation && hb.owner == lease.owner && hb.hb_seq > last_seen_hb_seq

`hb_seq` values are only comparable under the same hb owner — which is
exactly why `last_seen_hb_owner` was stored — but the stored owner was
never read (the half-finished check this fixes). The hazard is reachable:
a deposed leader's heartbeat thread keeps pulsing (`i_am_leader` resets
only on its next round) with `owner = itself`, and `pulseHeartbeat`'s
losing CAS writes vanish silently, so a zombie old leader can keep
`gc/hb.owner` pointing at itself while the live new leader is pulsing
too. On both of a follower's ticks `hb.owner != lease.owner` then made
`hb_alive` false, and with the new leader's lease tuple frozen mid-round
the follower stole the lease from a live, pulsing leader.

The gate is now observation-symmetric with the frozen-lease check: ANY
movement of the observed `(owner, hb_seq)` pair between the contender's
two ticks is proof of life, and `hb_seq` is compared only under the same
remembered hb owner (an owner change re-arms the window). Liveness is
preserved — only leaders pulse and a deposed leader stops within one round
interval, so a genuinely dead leader still gets stolen from one window
later at most.

Failing-first regression test
`CasGcLease.StaleOwnerHeartbeatDoesNotEnableFalseSteal` reproduces the
zombie-masking interleaving deterministically (steal confirmed on the old
predicate) and pins both the no-false-steal and the still-steals-when-
frozen halves.

Validated: full corrected CAS gtest battery 906/907 — the one failure is
the pre-existing tracked `RefWriterRecoverySeal` F3-1a fragility
(`docs/superpowers/cas/BACKLOG.md`), unrelated to the lease protocol.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: filimonov

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.cpp

@@ -2322,9 +2322,18 @@ bool Gc::acquireOrRenewLease(GcState & state, Token & state_token, bool allow_st
         GcHeartbeat hb;
         if (const auto hb_got = store->backend().get(store->layout().gcHbKey()))
             hb = decodeGcHeartbeat(hb_got->bytes);
+        /// Observation-based heartbeat liveness, symmetric with the frozen-lease-tuple check below:
+        /// ANY movement of the observed (owner, hb_seq) pair between this contender's two ticks is
+        /// proof of life, and `hb_seq` values are comparable only under the SAME remembered hb owner.
+        /// Deliberately NOT compared against `current.lease.owner`: a deposed leader's heartbeat
+        /// thread keeps pulsing (with `owner = itself`) until its next round resets `i_am_leader`,
+        /// and its writes can race out the live new leader's pulses — an hb pair that keeps moving
+        /// under the OLD owner's name must still read as "alive", or a live, pulsing new leader gets
+        /// its lease stolen. An hb owner change re-arms the window (this tick's pair is remembered
+        /// below); a steal happens only once the lease tuple AND the hb pair are both frozen across
+        /// a full window.
         const bool hb_alive = has_observation
-            && hb.owner == current.lease.owner
-            && hb.hb_seq > last_seen_hb_seq;
+            && (hb.owner != last_seen_hb_owner || hb.hb_seq > last_seen_hb_seq);
 
         const bool incumbent_renewed = !has_observation
             || current.lease.owner != last_seen_owner

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/Gc/CasGc.h

@@ -374,7 +374,9 @@ class Gc
     bool has_observation = false;
     UInt128 last_seen_owner{};
     uint64_t last_seen_seq = 0;
-    /// Heartbeat observed alongside the lease (gates the steal).
+    /// Heartbeat pair observed alongside the lease (gates the steal): a steal requires the lease
+    /// tuple AND this (owner, hb_seq) pair to be frozen across a full window. `hb_seq` is compared
+    /// only when the remembered hb owner matches; an owner change counts as movement (alive).
     UInt128 last_seen_hb_owner{};
     uint64_t last_seen_hb_seq = 0;
 

src/Disks/tests/gtest_cas_gc_round.cpp

@@ -46,6 +46,7 @@ namespace
 const UInt128 kGc = hexToU128("00000000000000000000000000000001");
 const UInt128 kGcA = hexToU128("0000000000000000000000000000000a");
 const UInt128 kGcB = hexToU128("0000000000000000000000000000000b");
+const UInt128 kGcC = hexToU128("0000000000000000000000000000000c");
 
 ManifestRef ref(uint64_t seq, uint64_t inst)
 {
@@ -329,6 +330,46 @@ TEST(CasGcLease, AcquireTimePulseProtectsNewLeadersFirstRound)
     EXPECT_EQ(readState(*b, *s).lease.owner, kGcA);       /// gc1 keeps the lease through its whole first round
 }
 
+TEST(CasGcLease, StaleOwnerHeartbeatDoesNotEnableFalseSteal)
+{
+    /// A deposed leader's heartbeat thread keeps pulsing until its next round notices the lost lease
+    /// (`i_am_leader` is only reset there), and `pulseHeartbeat` stamps `owner = self` while a losing
+    /// CAS write silently vanishes — so a zombie old leader can keep `gc/hb.owner` pointing at ITSELF
+    /// even while the live new leader is pulsing too. The liveness gate must therefore treat ANY
+    /// movement of the observed (owner, hb_seq) pair between a follower's two ticks as "someone is
+    /// alive": comparing hb_seq is only meaningful against the SAME remembered hb owner. The old
+    /// predicate compared `hb.owner` with the LEASE owner instead, so a zombie-owned hb read as
+    /// "not the leader's heartbeat" on both ticks and a live, pulsing new leader got its lease stolen.
+    std::shared_ptr<InMemoryBackend> b;
+    auto s = openTestPool(b);
+    Gc gc1(s, kGcA);
+    Gc gc2(s, kGcB);
+    Gc gc3(s, kGcC);
+
+    /// gc1 leads, beats, then dies mid-round; gc2 legitimately steals the lease.
+    ASSERT_TRUE(gc1.runRegularRound().acquired_lease);
+    Gc::pulseHeartbeat(*s, kGcA);
+    EXPECT_FALSE(gc2.runRegularRound().acquired_lease);   /// obs #1 of gc1's frozen tuple
+    EXPECT_TRUE(gc2.runRegularRound().acquired_lease);    /// obs #2: frozen lease + frozen hb => steal
+    ASSERT_EQ(readState(*b, *s).lease.owner, kGcB);
+
+    /// gc2 is now mid-long-round (lease tuple frozen) and PULSING — but gc1's zombie heartbeat
+    /// thread interleaves after every gc2 pulse, so the follower gc3 only ever OBSERVES gc1-owned
+    /// heartbeats. The pair keeps moving, which is proof of life.
+    Gc::pulseHeartbeat(*s, kGcB);
+    Gc::pulseHeartbeat(*s, kGcA);                          /// zombie masks gc2's pulse
+    EXPECT_FALSE(gc3.runRegularRound().acquired_lease);   /// obs #1: records (hb owner=A, seq)
+    Gc::pulseHeartbeat(*s, kGcB);
+    Gc::pulseHeartbeat(*s, kGcA);                          /// zombie masks again
+    EXPECT_FALSE(gc3.runRegularRound().acquired_lease);   /// obs #2: hb pair MOVED => alive => NO steal
+    EXPECT_EQ(readState(*b, *s).lease.owner, kGcB);       /// the live leader keeps its lease
+
+    /// Liveness is preserved: once everything genuinely freezes (gc2 dead, zombie gone), the next
+    /// tick completes the window — obs #2 above already re-armed on the now-frozen (lease, hb) pair.
+    EXPECT_TRUE(gc3.runRegularRound().acquired_lease);    /// still frozen a full tick later => steal
+    EXPECT_EQ(readState(*b, *s).lease.owner, kGcC);
+}
+
 TEST(CasGcLease, FailoverStealOnceHeartbeatStops)
 {
     /// B160: once the incumbent stops heartbeating (it died), a follower observing the now-frozen