fix(server-ng): stop redundant partition rebuilds resetting offsets

core/partitions/src/iggy_partition.rs

@@ -770,7 +770,24 @@ where
         self.offset.store(recovered_end, Ordering::Release);
         self.dirty_offset.store(recovered_end, Ordering::Relaxed);
         self.should_increment_offset = true;
-        self.stats.set_current_offset(recovered_end);
+    }
+
+    /// Copy this incarnation's offset counter into the shared
+    /// [`PartitionStats`], making it the value readers (offset validation,
+    /// `get_topic`, `get_stats`) see.
+    ///
+    /// Called from [`IggyPartitions::insert`](crate::IggyPartitions::insert)
+    /// only: when the instance BECOMES the addressable one, never while
+    /// building it. The stats registry keys on the namespace, not the
+    /// incarnation, so every build of a namespace holds the same `Arc` as
+    /// whatever is already serving it -- and a build is not guaranteed to be
+    /// adopted. Seeding from the build instead leaves a zeroed `current_offset`
+    /// on the live incarnation, which then rejects every
+    /// `store_consumer_offset` above 0 with `InvalidOffset` until the next send
+    /// re-seeds it.
+    pub(crate) fn publish_current_offset(&self) {
+        self.stats
+            .set_current_offset(self.offset.load(Ordering::Acquire));
     }
 
     /// The next message offset this replica will mint, `0` while the offset

core/partitions/src/iggy_partitions.rs

@@ -200,6 +200,12 @@ where
 
     /// Insert a new partition and return its local index.
     ///
+    /// Insertion is the moment a build becomes the addressable incarnation,
+    /// so this is also where its offset counter is published into the shared
+    /// `PartitionStats` ([`IggyPartition::publish_current_offset`]). No
+    /// earlier point is safe: a build that is never adopted must leave the
+    /// live incarnation's counters alone.
+    ///
     /// # Safety discipline (compiler cannot enforce)
     ///
     /// Must only be called from the shard's pump task (i.e. inside
@@ -220,6 +226,7 @@ where
             0,
             "IggyPartitions::insert while a with_partition borrow is live"
         );
+        partition.publish_current_offset();
         // Safety: pump-only invariant, caller responsibility.
         let partitions = unsafe { &mut *self.partitions.get() };
         let local_idx = LocalIdx::new(partitions.len());

core/server-ng/src/bootstrap.rs

@@ -2413,7 +2413,6 @@ async fn load_partition(
     partition.offset.store(counter, Ordering::Release);
     partition.dirty_offset.store(counter, Ordering::Relaxed);
     partition.should_increment_offset = current_offset.is_some();
-    partition.stats.set_current_offset(counter);
     // The durable frontier is a LOWER BOUND on top of what the segments proved:
     // it is the only carrier left when the segments that named the frontier are
     // gone (an all-GC'd origin's install, a crash inside the swap window), and

core/server-ng/src/partition_helpers.rs

@@ -694,7 +694,6 @@ pub async fn build_partition_fresh(
     partition.offset.store(0, Ordering::Release);
     partition.dirty_offset.store(0, Ordering::Relaxed);
     partition.should_increment_offset = false;
-    partition.stats.set_current_offset(0);
     debug_assert!(
         !partition.log.has_segments(),
         "fresh partition must not carry recovered segments"