[client] pqkem: keep rotating after a rotation times out

When every offer of a data-path rotation was lost, the initiator dropped
the exchange and logged that it would retry next cycle. It never did: a
rekey only chains a new rotation from an exchange awaiting it, and none was
left. Rotation stopped until the next Signal re-bootstrap (a lazy wake, or
never on a long-lived connection), and because no further attempt ran, the
consecutive-failure count could not reach the threshold that hands recovery
to the host either (reported in the security review of #7098).

On a rotation timeout, hand the chain back to the exchange the rotation
acknowledged. The next rekey then starts a fresh rotation from it: if the
offer was simply lost both peers still hold that PSK and the rotation
converges, and if the responder had already moved on, the attempts keep
failing until the threshold triggers a re-bootstrap. A timed-out bootstrap
is still dropped, as it has nothing to fall back to.

client/internal/pqkem/convergence.go

@@ -72,6 +72,7 @@ func (m *Manager) startExchangeLocked(remoteID RemoteID, viaSignal bool, ackID E
 		lastSent:  raw,
 		initiator: init,
 		viaSignal: viaSignal,
+		acks:      ackID,
 	}
 
 	m.wait.Add(1)
@@ -325,6 +326,12 @@ func (m *Manager) initiatorLoop(ctx context.Context, remoteID RemoteID, id Excha
 			case stateAwaitingAnswer:
 				if attempts >= m.maxRetries {
 					delete(m.exchanges, remoteID)
+					// A timed-out rotation hands the chain back to the exchange it
+					// acknowledged, so the next data-path rekey starts a new rotation.
+					// Without it rotation stops until the next Signal re-bootstrap.
+					if ex.acks != (ExchangeID{}) {
+						m.exchanges[remoteID] = &exchangeCtl{id: ex.acks, state: stateAwaitingRekey}
+					}
 					initial := !m.established[remoteID]
 					fail := m.registerFailureLocked(remoteID)
 					m.mu.Unlock()

client/internal/pqkem/manager.go

@@ -87,6 +87,7 @@ type exchangeCtl struct {
 	lastSent  []byte
 	initiator *Initiator
 	viaSignal bool
+	acks      ExchangeID // exchange this one acknowledges; zero for a bootstrap
 }
 
 // Manager is the stateful orchestrator — the analogue of go-rosenpass's Server. It

client/internal/pqkem/rotation_test.go

@@ -0,0 +1,69 @@
+package pqkem
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestManager_RotationResumesAfterTimeout covers a rotation whose offers are all lost.
+// The timeout must leave the rotation chain usable: the next data-path rekey has to start
+// a new rotation. Deleting the exchange on timeout stopped rotation for good, since a
+// rekey only chains from an exchange awaiting it, and the failure count never reached the
+// threshold that triggers a Signal re-bootstrap, despite the log promising a retry.
+func TestManager_RotationResumesAfterTimeout(t *testing.T) {
+	dA, dB, wgA, wgB, lbB := pair(t) // dB ("bbbb") initiator
+	defer dA.Stop()
+	defer dB.Stop()
+	dB.retryInterval = 10 * time.Millisecond
+	dB.maxRetries = 2
+
+	bootstrap(t, dA, dB)
+	boot := wgB.psk("aaaa")
+
+	// Lose every rotation offer until the exchange times out.
+	lbB.drop.Store(true)
+	dB.OnDataPathRekeyed("aaaa", 0)
+	require.Eventually(t, func() bool {
+		dB.mu.Lock()
+		defer dB.mu.Unlock()
+		ex := dB.exchanges["aaaa"]
+		return ex == nil || ex.state != stateAwaitingAnswer
+	}, time.Second, 5*time.Millisecond, "the lost rotation must time out")
+	lbB.drop.Store(false)
+
+	dB.OnDataPathRekeyed("aaaa", 0)
+	require.Eventually(t, func() bool {
+		return wgB.psk("aaaa") != boot && wgA.psk("bbbb") == wgB.psk("aaaa")
+	}, time.Second, 5*time.Millisecond, "the next rekey must start a rotation that converges")
+	assert.Empty(t, wgB.failed, "a single lost rotation must not escalate to a re-bootstrap")
+}
+
+// TestManager_RepeatedRotationTimeoutsEscalate verifies the failure policy can complete:
+// consecutive lost rotations must reach the rekey-failure threshold and hand recovery to
+// the host. With rotation stopping after the first timeout the threshold was unreachable.
+func TestManager_RepeatedRotationTimeoutsEscalate(t *testing.T) {
+	dA, dB, _, wgB, lbB := pair(t) // dB ("bbbb") initiator
+	defer dA.Stop()
+	defer dB.Stop()
+	dB.retryInterval = 10 * time.Millisecond
+	dB.maxRetries = 2
+
+	bootstrap(t, dA, dB)
+	lbB.drop.Store(true)
+
+	for i := 0; i < dB.maxRekeyFailures; i++ {
+		dB.OnDataPathRekeyed("aaaa", 0)
+		require.Eventually(t, func() bool {
+			dB.mu.Lock()
+			defer dB.mu.Unlock()
+			ex := dB.exchanges["aaaa"]
+			return ex == nil || ex.state != stateAwaitingAnswer
+		}, time.Second, 5*time.Millisecond, "rotation %d must time out", i+1)
+	}
+
+	assert.Eventually(t, func() bool { return failedCount(wgB) == 1 }, time.Second, 5*time.Millisecond,
+		"consecutive lost rotations must escalate to a rekey failure")
+}