Fix dead shard recovery and subscription replay on reconnect

Replace initial_subscriptions with a subscription_provider callback
that reads current subs from manager state on every connect/reconnect.
Wrap connection run in a panic retry loop (max 3 attempts) that
recreates the channel pair and updates state before retrying.

Remove redundant subscription sends from start() since the provider
now handles replay automatically. Add pre-commit hook for local CI.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: KaboomFox

.githooks/pre-commit

@@ -0,0 +1,71 @@
+#!/usr/bin/env bash
+set -e
+
+# Pre-commit hook: runs the same checks as CI to catch issues before pushing.
+# Install: git config core.hooksPath .githooks
+
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+YELLOW='\033[0;33m'
+BOLD='\033[1m'
+NC='\033[0m'
+
+failed=0
+
+step() {
+    printf "${BOLD}[pre-commit]${NC} %s... " "$1"
+}
+
+pass() {
+    printf "${GREEN}ok${NC} (%.1fs)\n" "$1"
+}
+
+fail() {
+    printf "${RED}FAILED${NC} (%.1fs)\n" "$1"
+    failed=1
+}
+
+# 1. Format check (fastest, catches trivial issues)
+step "cargo fmt --check"
+start=$(date +%s.%N)
+if cargo fmt --all -- --check >/dev/null 2>&1; then
+    pass "$(echo "$(date +%s.%N) - $start" | bc)"
+else
+    fail "$(echo "$(date +%s.%N) - $start" | bc)"
+    echo ""
+    printf "  ${YELLOW}Fix: cargo fmt --all${NC}\n"
+    echo ""
+fi
+
+# 2. Clippy (catches warnings/lint issues)
+step "cargo clippy"
+start=$(date +%s.%N)
+if cargo clippy --all-features -- -D warnings 2>&1 | tail -20 > /tmp/pre-commit-clippy.log 2>&1; then
+    pass "$(echo "$(date +%s.%N) - $start" | bc)"
+else
+    fail "$(echo "$(date +%s.%N) - $start" | bc)"
+    echo ""
+    cat /tmp/pre-commit-clippy.log
+    echo ""
+fi
+
+# 3. Tests (catches regressions)
+step "cargo test"
+start=$(date +%s.%N)
+if cargo test --all-features 2>&1 | tail -20 > /tmp/pre-commit-test.log 2>&1; then
+    pass "$(echo "$(date +%s.%N) - $start" | bc)"
+else
+    fail "$(echo "$(date +%s.%N) - $start" | bc)"
+    echo ""
+    cat /tmp/pre-commit-test.log
+    echo ""
+fi
+
+# Result
+echo ""
+if [ "$failed" -ne 0 ]; then
+    printf "${RED}${BOLD}Pre-commit checks failed.${NC} Fix the issues above or skip with --no-verify\n"
+    exit 1
+else
+    printf "${GREEN}${BOLD}All pre-commit checks passed.${NC}\n"
+fi

src/connection.rs

@@ -19,6 +19,10 @@ use tokio_tungstenite::{
 use tracing::{debug, error, info, trace, warn};
 use url::Url;
 
+/// A callback that provides the current subscriptions for a shard.
+/// Called on every connect/reconnect to get the latest subscription state.
+pub(crate) type SubscriptionProvider<S> = Arc<dyn Fn() -> Vec<S> + Send + Sync>;
+
 /// Commands that can be sent to a connection
 #[derive(Debug)]
 pub enum ConnectionCommand {
@@ -75,8 +79,8 @@ pub struct Connection<H: WebSocketHandler> {
     command_rx: mpsc::Receiver<ConnectionCommand>,
     /// Optional channel to signal when connection is ready
     ready_tx: Option<oneshot::Sender<()>>,
-    /// Explicit subscriptions for this shard (used during hot switchover)
-    initial_subscriptions: Option<Vec<H::Subscription>>,
+    /// Provides current subscriptions for this shard on connect/reconnect
+    subscription_provider: SubscriptionProvider<H::Subscription>,
     /// Optional channel to request hot switchover from manager
     switchover_tx: Option<mpsc::Sender<usize>>,
 }
@@ -93,6 +97,7 @@ impl<H: WebSocketHandler> Connection<H> {
         metrics: Arc<Metrics>,
         command_rx: mpsc::Receiver<ConnectionCommand>,
         switchover_tx: mpsc::Sender<usize>,
+        subscription_provider: SubscriptionProvider<H::Subscription>,
     ) -> Self {
         Self {
             shard_id,
@@ -103,7 +108,7 @@ impl<H: WebSocketHandler> Connection<H> {
             metrics,
             command_rx,
             ready_tx: None,
-            initial_subscriptions: None,
+            subscription_provider,
             switchover_tx: Some(switchover_tx),
         }
     }
@@ -119,7 +124,7 @@ impl<H: WebSocketHandler> Connection<H> {
         metrics: Arc<Metrics>,
         command_rx: mpsc::Receiver<ConnectionCommand>,
         ready_tx: oneshot::Sender<()>,
-        subscriptions: Vec<H::Subscription>,
+        subscription_provider: SubscriptionProvider<H::Subscription>,
         switchover_tx: mpsc::Sender<usize>,
     ) -> Self {
         Self {
@@ -131,7 +136,7 @@ impl<H: WebSocketHandler> Connection<H> {
             metrics,
             command_rx,
             ready_tx: Some(ready_tx),
-            initial_subscriptions: Some(subscriptions),
+            subscription_provider,
             switchover_tx: Some(switchover_tx),
         }
     }
@@ -391,18 +396,17 @@ impl<H: WebSocketHandler> Connection<H> {
             self.metrics.record_message_sent();
         }
 
-        // If we have explicit subscriptions (hot switchover), send subscription message
-        if let Some(ref subs) = self.initial_subscriptions {
-            if !subs.is_empty() {
-                if let Some(msg) = self.handler.subscription_message(subs) {
-                    write.send(msg).await?;
-                    self.metrics.record_message_sent();
-                    info!(
-                        "[SHARD-{}] Sent {} subscriptions from hot switchover",
-                        self.shard_id,
-                        subs.len()
-                    );
-                }
+        // Replay current subscriptions from provider (works for both reconnect and hot switchover)
+        let subs = (self.subscription_provider)();
+        if !subs.is_empty() {
+            if let Some(msg) = self.handler.subscription_message(&subs) {
+                write.send(msg).await?;
+                self.metrics.record_message_sent();
+                info!(
+                    "[SHARD-{}] Replayed {} subscriptions",
+                    self.shard_id,
+                    subs.len()
+                );
             }
         }