fix(pubsub): wake reconciliation only for a full queue; run the new tests in CI

A queued Subscribe is processed in order, and a closed queue means the
subscriber restarts and re-subscribes the whole desired map. Only a
subscribe hint dropped by a full queue needs the reconciliation wake-up,
so first retains no longer each make the subscriber rescan the map.

The new relay lifecycle tests and the full-queue pubsub test join the
infra-free unit lanes (Justfile and scripts/run-tests.sh). The Redis
subscriber test runs in a Redis-backed backend-integration step, so
buzz-pubsub joins that archive.

Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Signed-off-by: Logan Johnson <loganj@squareup.com>

.github/workflows/_ci-relay.yml

@@ -121,6 +121,7 @@ jobs:
           cargo nextest archive \
             --cargo-profile ci \
             -p buzz-db \
+            -p buzz-pubsub \
             -p buzz-relay \
             -p buzz-relay-mesh \
             -p buzz-test-client \
@@ -583,6 +584,13 @@ jobs:
             --archive-file target/ci/backend-integration-tests.tar.zst \
             -E 'package(buzz-relay-mesh) and test(/^registry::tests::external_infra_redis::/)' \
             --run-ignored ignored-only
+      - name: Pubsub subscriber reconciliation regression
+        # Runs the live Redis subscriber; the test dials redis://127.0.0.1:6379.
+        run: |
+          cargo nextest run \
+            --archive-file target/ci/backend-integration-tests.tar.zst \
+            -E 'package(buzz-pubsub) and test(=tests::live_subscriber_recovers_a_dropped_subscribe_hint)' \
+            --run-ignored ignored-only
       - name: Workflow message provenance unit tests
         # The relay's workflow_sink suite is not selected by the infra-free
         # unit job. Its ignored database cases run in the isolated PostgreSQL

crates/buzz-pubsub/src/lib.rs

@@ -259,16 +259,19 @@ impl PubSubManager {
         };
 
         if should_subscribe {
-            // The wake-up is lossless and coalesced: if the bounded command
-            // queue is full, the subscriber still reconciles the whole
-            // desired-topic map without waiting for a Redis reconnect.
-            self.subscription_changed.notify_one();
             match self
                 .subscription_tx
                 .try_send(subscriber::SubscriptionCommand::Subscribe(topic_key))
             {
+                // A queued command subscribes in order; a closed queue means
+                // the subscriber is gone, and a restart re-subscribes the
+                // whole desired-topic map.
                 Ok(()) | Err(mpsc::error::TrySendError::Closed(_)) => {}
                 Err(mpsc::error::TrySendError::Full(_)) => {
+                    // The wake-up is lossless and coalesced: the subscriber
+                    // reconciles the desired-topic map on the live link
+                    // instead of waiting for a Redis reconnect.
+                    self.subscription_changed.notify_one();
                     tracing::warn!(
                         ?topic_key,
                         "pubsub subscription command queue full; reconciliation scheduled"
@@ -730,8 +733,11 @@ mod tests {
     /// A retain past a full command queue neither waits for the subscriber
     /// (callers hold a lifecycle lock) nor loses the topic: it leaves a
     /// reconciliation wake-up, and the topic is in the desired map the
-    /// subscriber re-reads. Mutation: `send().await` instead of `try_send`
-    /// hangs at the timeout below; no `notify_one` fails the wake-up check.
+    /// subscriber re-reads. Retains that fit in the queue leave no wake-up,
+    /// so first retains do not each make the subscriber rescan the map.
+    /// Mutation: `send().await` instead of `try_send` hangs at the timeout
+    /// below; no `notify_one` fails the wake-up check; `notify_one` on every
+    /// first retain fails the no-wake-up check.
     #[tokio::test]
     async fn queue_full_retain_does_not_wait_and_schedules_reconciliation() {
         let manager = make_manager().await;
@@ -741,8 +747,16 @@ mod tests {
                 .retain_topic(&filler, EventTopic::Channel(Uuid::from_u128(id as u128)))
                 .await;
         }
-        // Drain the wake-ups the fillers left (coalesced into one permit).
-        manager.subscription_changed.notified().await;
+        // Retains that fit in the queue leave no reconciliation wake-up.
+        assert!(
+            tokio::time::timeout(
+                Duration::from_millis(10),
+                manager.subscription_changed.notified()
+            )
+            .await
+            .is_err(),
+            "a queued subscribe needs no reconciliation"
+        );
 
         let dropped_ctx = ctx(0xbbbb, "b.example");
         let dropped = EventTopic::Channel(Uuid::from_u128(0xcccc));

scripts/run-tests.sh

@@ -103,6 +103,12 @@ run_unit_tests() {
   run_test_step "buzz-pubsub connection tests" \
     cargo test -p buzz-pubsub --lib connection::tests:: -- --nocapture
 
+  # A retain past a full subscriber command queue (no Redis dialed;
+  # infra-free). Mirrors `just test-unit`.
+  run_test_step "buzz-pubsub full-queue retain test" \
+    cargo test -p buzz-pubsub --lib -- --exact --nocapture \
+      tests::queue_full_retain_does_not_wait_and_schedules_reconciliation
+
   run_test_step "buzz-voice tests" \
     cargo test -p buzz-voice --lib -- --nocapture
 
@@ -295,6 +301,10 @@ run_unit_tests() {
       handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner \
       handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim \
       handlers::req::tests::revoke_then_replacement_keeps_replacement_whole \
+      handlers::req::tests::revoke_of_a_replaced_holder_spares_the_pending_replacement \
+      handlers::req::tests::superseded_or_closed_requests_cannot_claim \
+      connection::tests::close_fences_a_req_whose_handler_has_not_run \
+      connection::tests::only_the_newest_same_id_req_answers \
       handlers::req::tests::claims_after_connection_cleanup_are_refused \
       handlers::req::tests::dropped_terminal_frame_cancels_connection \
       handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection \