Fix divergenceEvent pattern propagation, flush close flag, and window boundary

Co-authored-by: bootjp <1306365+bootjp@users.noreply.github.com>

Author: Copilot

proxy/compare.go

@@ -43,6 +43,7 @@ func (k DivergenceKind) String() string {
 type Divergence struct {
 	Command    string
 	Key        string
+	Pattern    string // non-empty for PSUBSCRIBE/pmessage divergences
 	Kind       DivergenceKind
 	Primary    any
 	Secondary  any

proxy/pubsub.go

@@ -720,12 +720,17 @@ func (s *pubsubSession) mirrorUnsub(names []string, isPattern bool) {
 // --- Helpers ---
 
 // flushOrClose flushes the detached connection. On error it closes the
-// connection so that commandLoop will observe a read failure and shut down.
-// Caller must hold s.writeMu.
+// connection and marks the session as closed so that background forwarding
+// loops exit promptly. Caller must hold s.writeMu.
 func (s *pubsubSession) flushOrClose() {
 	if err := s.dconn.Flush(); err != nil {
 		s.logger.Warn("failed to flush to client; closing connection", "err", err)
 		_ = s.dconn.Close()
+
+		// Mark the session as closed so background forwarding loops can exit promptly.
+		s.mu.Lock()
+		s.closed = true
+		s.mu.Unlock()
 	}
 }
 

proxy/sentry.go

@@ -91,6 +91,9 @@ func (r *SentryReporter) CaptureDivergence(div Divergence) {
 		// Omit raw key from Sentry tags to avoid leaking sensitive data;
 		// only send a truncated form as an extra for debugging.
 		scope.SetExtra("key", truncateValue(div.Key))
+		if div.Pattern != "" {
+			scope.SetExtra("pattern", truncateValue(div.Pattern))
+		}
 		scope.SetExtra("primary", truncateValue(div.Primary))
 		scope.SetExtra("secondary", truncateValue(div.Secondary))
 		scope.SetFingerprint([]string{"divergence", div.Kind.String(), div.Command})

proxy/shadow_pubsub.go

@@ -50,6 +50,7 @@ type pendingMsg struct {
 type divergenceEvent struct {
 	channel   string
 	payload   string
+	pattern   string // non-empty for PSUBSCRIBE/pmessage deliveries
 	kind      DivergenceKind
 	isPattern bool // true if this originated from a PSUBSCRIBE
 }
@@ -233,7 +234,7 @@ func (sp *shadowPubSub) matchSecondary(msg *redis.Message) {
 	now := sp.nowFunc()
 	if entries, ok := sp.pending[key]; ok && len(entries) > 0 {
 		oldest := entries[0]
-		if now.Sub(oldest.timestamp) < sp.window {
+		if now.Sub(oldest.timestamp) <= sp.window {
 			// Match found within window — remove the oldest pending primary message.
 			if len(entries) == 1 {
 				delete(sp.pending, key)
@@ -308,7 +309,7 @@ func (sp *shadowPubSub) reconcilePrimaries(now time.Time, secBuf map[msgKey][]se
 			if now.Sub(e.timestamp) >= sp.window {
 				// Primary has expired — report as divergence regardless of any buffered
 				// secondaries. A late secondary must not suppress a window violation.
-				out = append(out, divergenceEvent{channel: e.channel, payload: e.payload, kind: DivDataMismatch, isPattern: e.pattern != ""})
+				out = append(out, divergenceEvent{channel: e.channel, payload: e.payload, pattern: e.pattern, kind: DivDataMismatch, isPattern: e.pattern != ""})
 				continue
 			}
 			if secs := secBuf[key]; len(secs) > 0 {
@@ -337,7 +338,7 @@ func sweepExpiredSecondaries(now time.Time, window time.Duration, secBuf map[msg
 		var remaining []secondaryPending
 		for _, sec := range secs {
 			if now.Sub(sec.timestamp) >= window {
-				out = append(out, divergenceEvent{channel: sec.channel, payload: sec.payload, kind: DivExtraData, isPattern: key.Pattern != ""})
+				out = append(out, divergenceEvent{channel: sec.channel, payload: sec.payload, pattern: key.Pattern, kind: DivExtraData, isPattern: key.Pattern != ""})
 			} else {
 				remaining = append(remaining, sec)
 			}
@@ -362,6 +363,7 @@ func (sp *shadowPubSub) sweepAll() {
 			divergences = append(divergences, divergenceEvent{
 				channel:   e.channel,
 				payload:   e.payload,
+				pattern:   e.pattern,
 				kind:      DivDataMismatch,
 				isPattern: e.pattern != "",
 			})
@@ -378,6 +380,7 @@ func (sp *shadowPubSub) sweepAll() {
 				divergences = append(divergences, divergenceEvent{
 					channel:   sec.channel,
 					payload:   sec.payload,
+					pattern:   key.Pattern,
 					kind:      DivExtraData,
 					isPattern: key.Pattern != "",
 				})
@@ -395,11 +398,15 @@ func (sp *shadowPubSub) sweepAll() {
 
 func (sp *shadowPubSub) reportDivergence(d divergenceEvent) {
 	sp.metrics.PubSubShadowDivergences.WithLabelValues(d.kind.String()).Inc()
-	sp.logger.Warn("pubsub shadow divergence",
+	logAttrs := []any{
 		"channel", truncateValue(d.channel),
 		"payload", truncateValue(d.payload),
 		"kind", d.kind.String(),
-	)
+	}
+	if d.pattern != "" {
+		logAttrs = append(logAttrs, "pattern", truncateValue(d.pattern))
+	}
+	sp.logger.Warn("pubsub shadow divergence", logAttrs...)
 
 	cmd := "SUBSCRIBE"
 	if d.isPattern {
@@ -418,6 +425,7 @@ func (sp *shadowPubSub) reportDivergence(d divergenceEvent) {
 	sp.sentry.CaptureDivergence(Divergence{
 		Command:    cmd,
 		Key:        d.channel,
+		Pattern:    d.pattern,
 		Kind:       d.kind,
 		Primary:    primary,
 		Secondary:  secondary,

proxy/shadow_pubsub_test.go

@@ -315,6 +315,7 @@ func TestShadowPubSub_SecondaryWithinWindowMatches(t *testing.T) {
 	assert.Equal(t, float64(0), extra, "within-window secondary match must suppress extra_data")
 }
 
+// TestShadowPubSub_ExpiredPrimaryThenLateSecondary verifies that a secondary
 // arriving after the comparison window does not suppress the already-expired
 // primary divergence. Both a DivDataMismatch (primary) and eventually a
 // DivExtraData (secondary) should be reported.