refactor!: SubscriberGroup wraps Subscriber, drop redundant Publisher::sequence

SubscriberGroup carried a byte-for-byte copy of Subscriber's fields and of try_recv, recv_with, recv_batch, pending, the counter getters, update_tracker, and Drop. All N logical subscribers already share one cursor, so the type is behaviourally a Subscriber plus a compile-time count. It becomes a newtype over Subscriber that delegates, keeping aligned_count() as the only group-specific method.

Two incidental improvements fall out: the group now uses Subscriber::recv's two-phase spin (64 bare iterations before the power-efficient wait) instead of its own single-phase loop, and Send is auto-derived rather than an unsafe impl.

Publisher::sequence returned self.seq exactly as published() does. Removed, with the lag-computation note folded into published(); its test duplicated published_count and is removed with it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: userFRM

src/channel/group.rs

@@ -2,44 +2,26 @@
 // SPDX-License-Identifier: MIT OR Apache-2.0
 
 use super::errors::TryRecvError;
+use super::subscriber::Subscriber;
 use crate::pod::Pod;
-use crate::ring::{Padded, RingIndex, SharedRing};
-use crate::slot::Slot;
 use crate::wait::WaitStrategy;
-use alloc::sync::Arc;
-use core::sync::atomic::{AtomicU64, Ordering};
 
 /// A group of `N` logical subscribers backed by a single ring read.
 ///
 /// All `N` logical subscribers share one cursor —
 /// [`try_recv`](SubscriberGroup::try_recv) performs **one** seqlock read
 /// and a single cursor increment, eliminating the N-element sweep loop.
+/// That makes the group behaviourally a single [`Subscriber`] carrying a
+/// compile-time count, which is exactly how it is implemented; `N` is
+/// reported by [`aligned_count`](SubscriberGroup::aligned_count).
 ///
 /// ```
 /// let (mut p, subs) = photon_ring::channel::<u64>(64);
 /// let mut group = subs.subscribe_group::<4>();
 /// p.publish(42);
 /// assert_eq!(group.try_recv(), Ok(42));
 /// ```
-pub struct SubscriberGroup<T: Pod, const N: usize> {
-    pub(super) ring: Arc<SharedRing<T>>,
-    /// Cached raw pointer to the slot array. Avoids Arc + Box deref on the
-    /// hot path. Valid for the lifetime of `ring` (the Arc keeps it alive).
-    pub(super) slots_ptr: *const Slot<T>,
-    /// Precomputed slot indexing (capacity, mask, reciprocal, pow2 flag).
-    pub(super) index: RingIndex,
-    /// Single cursor shared by all `N` logical subscribers.
-    pub(super) cursor: u64,
-    /// Cumulative messages skipped due to lag.
-    pub(super) total_lagged: u64,
-    /// Cumulative messages successfully received.
-    pub(super) total_received: u64,
-    /// Per-group cursor tracker for backpressure. `None` on regular
-    /// (lossy) channels — zero overhead.
-    pub(super) tracker: Option<Arc<Padded<AtomicU64>>>,
-}
-
-unsafe impl<T: Pod, const N: usize> Send for SubscriberGroup<T, N> {}
+pub struct SubscriberGroup<T: Pod, const N: usize>(pub(super) Subscriber<T>);
 
 impl<T: Pod, const N: usize> SubscriberGroup<T, N> {
     /// Try to receive the next message for the group.
@@ -48,74 +30,21 @@ impl<T: Pod, const N: usize> SubscriberGroup<T, N> {
     /// N-element sweep needed since all logical subscribers share one cursor.
     #[inline]
     pub fn try_recv(&mut self) -> Result<T, TryRecvError> {
-        let cur = self.cursor;
-        // SAFETY: slots_ptr is valid for the lifetime of self.ring (Arc-owned).
-        let slot = unsafe { &*self.slots_ptr.add(self.index.slot(cur)) };
-        let expected = cur * 2 + 2;
-
-        match slot.try_read(cur) {
-            Ok(Some(value)) => {
-                self.cursor = cur + 1;
-                self.total_received += 1;
-                self.update_tracker();
-                Ok(value)
-            }
-            Ok(None) => Err(TryRecvError::Empty),
-            Err(actual_stamp) => {
-                if actual_stamp & 1 != 0 || actual_stamp < expected {
-                    return Err(TryRecvError::Empty);
-                }
-                // Lagged — recompute from head cursor
-                let head = self.ring.cursor.0.load(Ordering::Acquire);
-                let cap = self.ring.capacity();
-                if head == u64::MAX || cur > head {
-                    return Err(TryRecvError::Empty);
-                }
-                if head >= cap {
-                    let oldest = head - cap + 1;
-                    if cur < oldest {
-                        let skipped = oldest - cur;
-                        self.cursor = oldest;
-                        self.total_lagged += skipped;
-                        self.update_tracker();
-                        return Err(TryRecvError::Lagged { skipped });
-                    }
-                }
-                Err(TryRecvError::Empty)
-            }
-        }
+        self.0.try_recv()
     }
 
     /// Spin until the next message is available.
     ///
-    /// On aarch64: uses SEVL + WFE for near-zero-power cache-line-event
-    /// wakeup. On x86: uses PAUSE (spin_loop hint).
+    /// Uses the same two-phase spin as [`Subscriber::recv`]: bare spin for
+    /// the first 64 iterations, then a power-efficient wait (`PAUSE` on x86,
+    /// `SEVL`/`WFE` on aarch64).
     #[inline]
     pub fn recv(&mut self) -> T {
-        #[cfg(target_arch = "aarch64")]
-        unsafe {
-            core::arch::asm!("sevl", options(nomem, nostack));
-        }
-        loop {
-            match self.try_recv() {
-                Ok(val) => return val,
-                Err(TryRecvError::Empty) => {
-                    #[cfg(target_arch = "aarch64")]
-                    unsafe {
-                        core::arch::asm!("wfe", options(nomem, nostack));
-                    }
-                    #[cfg(not(target_arch = "aarch64"))]
-                    core::hint::spin_loop();
-                }
-                Err(TryRecvError::Lagged { .. }) => {}
-            }
-        }
+        self.0.recv()
     }
 
     /// Block until the next message using the given [`WaitStrategy`].
     ///
-    /// Like [`Subscriber::recv_with`], but for the grouped fast path.
-    ///
     /// # Example
     /// ```
     /// use photon_ring::{channel, WaitStrategy};
@@ -127,49 +56,7 @@ impl<T: Pod, const N: usize> SubscriberGroup<T, N> {
     /// ```
     #[inline]
     pub fn recv_with(&mut self, strategy: WaitStrategy) -> T {
-        let cur = self.cursor;
-        let slot = unsafe { &*self.slots_ptr.add(self.index.slot(cur)) };
-        let expected = cur * 2 + 2;
-        let mut iter: u32 = 0;
-        loop {
-            match slot.try_read(cur) {
-                Ok(Some(value)) => {
-                    self.cursor = cur + 1;
-                    self.total_received += 1;
-                    self.update_tracker();
-                    return value;
-                }
-                Ok(None) => {
-                    strategy.wait(iter);
-                    iter = iter.saturating_add(1);
-                }
-                Err(stamp) => {
-                    if stamp >= expected {
-                        return self.recv_with_slow(strategy);
-                    }
-                    strategy.wait(iter);
-                    iter = iter.saturating_add(1);
-                }
-            }
-        }
-    }
-
-    #[cold]
-    #[inline(never)]
-    fn recv_with_slow(&mut self, strategy: WaitStrategy) -> T {
-        let mut iter: u32 = 0;
-        loop {
-            match self.try_recv() {
-                Ok(val) => return val,
-                Err(TryRecvError::Empty) => {
-                    strategy.wait(iter);
-                    iter = iter.saturating_add(1);
-                }
-                Err(TryRecvError::Lagged { .. }) => {
-                    iter = 0;
-                }
-            }
-        }
+        self.0.recv_with(strategy)
     }
 
     /// How many of the `N` logical subscribers are aligned.
@@ -184,37 +71,26 @@ impl<T: Pod, const N: usize> SubscriberGroup<T, N> {
     /// Number of messages available to read (capped at ring capacity).
     #[inline]
     pub fn pending(&self) -> u64 {
-        let head = self.ring.cursor.0.load(Ordering::Acquire);
-        if head == u64::MAX || self.cursor > head {
-            0
-        } else {
-            let raw = head - self.cursor + 1;
-            raw.min(self.ring.capacity())
-        }
+        self.0.pending()
     }
 
     /// Total messages successfully received by this group.
     #[inline]
     pub fn total_received(&self) -> u64 {
-        self.total_received
+        self.0.total_received()
     }
 
     /// Total messages lost due to lag (group fell behind the ring).
     #[inline]
     pub fn total_lagged(&self) -> u64 {
-        self.total_lagged
+        self.0.total_lagged()
     }
 
     /// Ratio of received to total (received + lagged). Returns 0.0 if no
     /// messages have been processed.
     #[inline]
     pub fn receive_ratio(&self) -> f64 {
-        let total = self.total_received + self.total_lagged;
-        if total == 0 {
-            0.0
-        } else {
-            self.total_received as f64 / total as f64
-        }
+        self.0.receive_ratio()
     }
 
     /// Receive up to `buf.len()` messages in a single call.
@@ -224,47 +100,6 @@ impl<T: Pod, const N: usize> SubscriberGroup<T, N> {
     /// advanced and filling continues from the oldest available message.
     #[inline]
     pub fn recv_batch(&mut self, buf: &mut [T]) -> usize {
-        let mut count = 0;
-        for slot in buf.iter_mut() {
-            match self.try_recv() {
-                Ok(value) => {
-                    *slot = value;
-                    count += 1;
-                }
-                Err(TryRecvError::Empty) => break,
-                Err(TryRecvError::Lagged { .. }) => {
-                    // Cursor was advanced — retry from oldest available.
-                    match self.try_recv() {
-                        Ok(value) => {
-                            *slot = value;
-                            count += 1;
-                        }
-                        Err(_) => break,
-                    }
-                }
-            }
-        }
-        count
-    }
-
-    /// Update the backpressure tracker to reflect the current cursor position.
-    /// No-op on regular (lossy) channels.
-    #[inline]
-    fn update_tracker(&self) {
-        if let Some(ref tracker) = self.tracker {
-            tracker.0.store(self.cursor, Ordering::Relaxed);
-        }
-    }
-}
-
-impl<T: Pod, const N: usize> Drop for SubscriberGroup<T, N> {
-    fn drop(&mut self) {
-        if let Some(ref tracker) = self.tracker {
-            if let Some(ref bp) = self.ring.backpressure {
-                let weak = Arc::downgrade(tracker);
-                let mut trackers = bp.trackers.lock();
-                trackers.retain(|t| !t.ptr_eq(&weak));
-            }
-        }
+        self.0.recv_batch(buf)
     }
 }

src/channel/publisher.rs

@@ -196,19 +196,14 @@ impl<T: Pod> Publisher<T> {
         }
     }
 
-    /// Number of messages published so far.
+    /// Number of messages published so far, which is also the next sequence
+    /// number to be written. Useful for computing lag:
+    /// `publisher.published() - subscriber.cursor`.
     #[inline]
     pub fn published(&self) -> u64 {
         self.seq
     }
 
-    /// Current sequence number (same as `published()`).
-    /// Useful for computing lag: `publisher.sequence() - subscriber.cursor`.
-    #[inline]
-    pub fn sequence(&self) -> u64 {
-        self.seq
-    }
-
     /// Ring capacity.
     #[inline]
     pub fn capacity(&self) -> u64 {

src/channel/subscribable.rs

@@ -58,20 +58,7 @@ impl<T: Pod> Subscribable<T> {
     /// Panics if `N` is 0.
     pub fn subscribe_group<const N: usize>(&self) -> SubscriberGroup<T, N> {
         assert!(N > 0, "SubscriberGroup requires at least 1 subscriber");
-        let head = self.ring.cursor.0.load(Ordering::Acquire);
-        let start = if head == u64::MAX { 0 } else { head + 1 };
-        let tracker = self.ring.register_tracker(start);
-        let slots_ptr = self.ring.slots_ptr();
-        let idx = self.ring.index;
-        SubscriberGroup {
-            ring: self.ring.clone(),
-            slots_ptr,
-            index: idx,
-            cursor: start,
-            total_lagged: 0,
-            total_received: 0,
-            tracker,
-        }
+        SubscriberGroup(self.subscribe())
     }
 
     /// Create a subscriber starting from the **oldest available** message

tests/correctness.rs

@@ -813,18 +813,6 @@ fn group_counters() {
     assert!(group.total_lagged() > 0);
 }
 
-#[test]
-fn publisher_sequence() {
-    let (mut p, _s) = channel::<u64>(8);
-    assert_eq!(p.sequence(), 0);
-    p.publish(1);
-    assert_eq!(p.sequence(), 1);
-    p.publish_batch(&[2, 3, 4]);
-    assert_eq!(p.sequence(), 4);
-    // sequence() == published()
-    assert_eq!(p.sequence(), p.published());
-}
-
 // -------------------------------------------------------------------------
 // Bug fix: publish() respects backpressure on bounded channels
 // -------------------------------------------------------------------------