Race fix

client/internal/pqkem/convergence.go

@@ -140,12 +140,12 @@ func (m *Manager) ackConverged(remoteID string, ackID ExchangeID) {
 	m.mu.Unlock()
 }
 
-// initiatorLoop enforces the convergence deadline and retransmits the initiator's
+// initiatorLoop enforces the offer->answer convergence deadline and retransmits the
-// outstanding data-path offer while awaiting the answer (a signalling-bootstrapped
+// initiator's outstanding data-path offer while awaiting the answer (a
-// offer is retransmitted by the host, so it is not resent here). It then waits,
+// signalling-bootstrapped offer is retransmitted by the host, so it is not resent
-// counting toward the deadline, in stateAwaitingRekey until OnDataPathRekeyed chains
+// here). Exhausting the deadline before the answer arrives is a failure. Once the
-// the next exchange (which supersedes and cancels this loop). Exhausting the deadline
+// answer is in (state past awaitingAnswer) the loop exits: the next rotation is driven
-// is a failure.
+// by OnDataPathRekeyed, and the idle wait for it has no deadline.
 func (m *Manager) initiatorLoop(ctx context.Context, remoteID string, id ExchangeID) {
 	defer m.wait.Done()
 	t := time.NewTicker(m.retryInterval)
@@ -183,20 +183,11 @@ func (m *Manager) initiatorLoop(ctx context.Context, remoteID string, id Exchang
 					}
 				}
 
-			case stateAwaitingRekey:
-				// Waiting for OnDataPathRekeyed to chain the next exchange; the
-				// deadline still applies (the data path may never adopt the new key).
-				if attempts >= m.maxRetries {
-					delete(m.exchanges, remoteID)
-					fail := m.registerFailureLocked(remoteID)
-					m.mu.Unlock()
-					m.raiseFailure(remoteID, fail)
-					return
-				}
-				attempts++
-				m.mu.Unlock()
-
 			default:
+				// Past awaiting the answer (converged) or superseded: the loop's job
+				// is done. The next rotation is driven externally by OnDataPathRekeyed,
+				// so there is no deadline while idle-waiting for it (that wait can be
+				// as long as the transport's natural rekey interval).
 				m.mu.Unlock()
 				return
 			}

client/internal/pqkem/convergence_test.go

@@ -11,10 +11,10 @@ import (
 // dropTransport is a pqkem.Transport that silently discards everything.
 type dropTransport struct{}
 
-func (dropTransport) Send(netip.AddrPort, []byte) error       { return nil }
+func (dropTransport) Send(netip.AddrPort, []byte) error { return nil }
-func (dropTransport) LocalPort() int                          { return 0 }
+func (dropTransport) LocalPort() int                    { return 0 }
-func (dropTransport) Run(func(netip.AddrPort, []byte))        {}
+func (dropTransport) Run(func(netip.AddrPort, []byte))  {}
-func (dropTransport) Close() error                            { return nil }
+func (dropTransport) Close() error                      { return nil }
 
 func failedCount(f *fakeWG) int {
 	f.mu.Lock()
@@ -44,6 +44,11 @@ func TestManager_RekeyToleratesKFailures(t *testing.T) {
 	defer dA.Stop()
 	defer dB.Stop()
 
+	// Tighten B's timings before any exchange loop spawns (the loop reads these
+	// fields, so writing them after a loop is running would race).
+	dB.retryInterval = 5 * time.Millisecond
+	dB.maxRetries = 2
+
 	// Establish: bootstrap + data-path-rekeyed so B becomes established and its data
 	// path is usable.
 	bootstrap(t, dA, dB)
@@ -51,9 +56,7 @@ func TestManager_RekeyToleratesKFailures(t *testing.T) {
 	dB.OnDataPathRekeyed("aaaa")
 	require.NotEqual(t, PSK{}, wgB.psk("aaaa"))
 
-	// Tighten timings and drop B's outbound so rekeys can no longer converge.
+	// Drop B's outbound so rekeys can no longer converge.
-	dB.retryInterval = 5 * time.Millisecond
-	dB.maxRetries = 2
 	lbB.drop.Store(true)
 
 	// K-1 data-path rekeys must NOT raise OnRekeyFailed.