Fixing flaky integration test: TestCrashAfterRejectDoesNotLoseMessages (#274)

## Summary
TestCrashAfterRejectDoesNotLoseMessages was flaky because it asserted
consumer lag immediately after Ack(), but Ack() only marks the delivery
state — watermark advancement (which updates offset_acked) is deferred
to the next poll loop tick. The test raced against the poll loop: if the
lag check ran before advanceWatermark, offset_acked was still stale and
lag was non-zero.

Fix by adding OnSignal to worker-2's queue and calling waitForSignal
after acks, ensuring the poll loop has run advanceWatermark before we
check lag. This matches the pattern already used by
TestWatermarkAdvancesContiguously.

## Test Plan
Ran the test 50 times to confirm it's no longer flaky:
```
./tool/bazel test //test/integration/extension/messagequeue/mysql:mysql_test --test_filter='TestSQLQueueIntegration/TestCrashAfterRejectDoesNotLoseMessages' --runs_per_test=50 --test_output=summary --jobs=4
     INFO: Found 1 test target...
     Target //test/integration/extension/messagequeue/mysql:mysql_test up-to-date:
       bazel-bin/test/integration/extension/messagequeue/mysql/mysql_test_/mysql_test
     INFO: Elapsed time: 205.664s, Critical Path: 20.62s
     INFO: 51 processes: 9 action cache hit, 1 internal, 50 darwin-sandbox.
     INFO: Build completed successfully, 51 total actions
     //test/integration/extension/messagequeue/mysql:mysql_test               PASSED in 20.6s
       Stats over 50 runs: max = 20.6s, min = 10.0s, avg = 16.0s, dev = 2.0s

     Executed 1 out of 1 test: 1 test passes.
```
## Issue
<!-- 
Link the issue here. 
- Use 'Closes #123' if this is the final fix. 
- Use 'Part of #123' or just '#123' if the feature is still in progress.
-->

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ubettigole

test/integration/extension/messagequeue/mysql/queue_test.go

@@ -2221,10 +2221,12 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRejectDoesNotLoseMessages() {
 
 	// Start worker-2 with same consumer group — it polls and finds msg-C
 	// after lease + visibility expire in the DB
+	signalCh := make(chan queueMySQL.HookSignal, 100)
 	q2, err := queueMySQL.NewQueue(queueMySQL.Params{
 		DB:           s.db,
 		Logger:       zaptest.NewLogger(t),
 		MetricsScope: tally.NoopScope,
+		OnSignal:     signalCh,
 	})
 	require.NoError(t, err)
 	defer q2.Close()
@@ -2243,6 +2245,9 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRejectDoesNotLoseMessages() {
 	require.NoError(t, delivery.Ack(s.ctx))
 	t.Logf("Worker-2 recovered msg-C (attempt=%d)", delivery.Attempt())
 
+	// Wait for the poll loop to advance the watermark after acking msg-C.
+	waitForSignal(t, signalCh, queueMySQL.SignalDeliveryCheck)
+
 	// Verify DLQ contains msg-B
 	dlqTopic := topic + subConfig.DLQ.TopicSuffix
 	dlqConfig := extqueue.DefaultSubscriptionConfig("worker-2", "crash-reject-cg")
@@ -2254,7 +2259,9 @@ func (s *SQLQueueIntegrationSuite) TestCrashAfterRejectDoesNotLoseMessages() {
 	assert.Equal(t, "msg-B", dlqDelivery.Message().ID, "msg-B should be in DLQ")
 	require.NoError(t, dlqDelivery.Ack(s.ctx))
 
-	// Verify consumer lag is 0
+	// Verify consumer lag is 0.
+	// Wait for the poll loop so advanceWatermark has run after all acks.
+	waitForSignal(t, signalCh, queueMySQL.SignalDeliveryCheck)
 	admin := queueAdmin.NewAdminStore(s.db)
 	lags, err := admin.ConsumerLag(s.ctx, topic)
 	require.NoError(t, err)