SVI/netbird
1
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-07-29 01:12:36 -04:00
Code Issues 3.0k Packages Projects Releases 10 Wiki Activity

Race fix

Browse Source
This commit is contained in:
riccardom
2026-07-25 12:12:32 +02:00
parent a9fb4e9f0a
commit d544bfa15e
2 changed files with 20 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
client/internal/pqkem/convergence.go
View File

@@ -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
// outstanding data-path offer while awaiting the answer (a signalling-bootstrapped
// offer is retransmitted by the host, so it is not resent here). It then waits,
// counting toward the deadline, in stateAwaitingRekey until OnDataPathRekeyed chains
// the next exchange (which supersedes and cancels this loop). Exhausting the deadline
// is a failure.
// initiatorLoop enforces the offer->answer convergence deadline and retransmits the
// initiator's outstanding data-path offer while awaiting the answer (a
// signalling-bootstrapped offer is retransmitted by the host, so it is not resent
// here). Exhausting the deadline before the answer arrives is a failure. Once the
// answer is in (state past awaitingAnswer) the loop exits: the next rotation is driven
// 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
}

17
client/internal/pqkem/convergence_test.go
View File

@@ -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) LocalPort() int { return 0 }
func (dropTransport) Run(func(netip.AddrPort, []byte)) {}
func (dropTransport) Close() error { return nil }
func (dropTransport) Send(netip.AddrPort, []byte) error { return nil }
func (dropTransport) LocalPort() int { return 0 }
func (dropTransport) Run(func(netip.AddrPort, []byte)) {}
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.
dB.retryInterval = 5 * time.Millisecond
dB.maxRetries = 2
// Drop B's outbound so rekeys can no longer converge.
lbB.drop.Store(true)
// K-1 data-path rekeys must NOT raise OnRekeyFailed.