mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-27 06:49:07 +01:00
## Describe your changes Restore the management gRPC client's backoff reset after a stream connects. This is a regression: PR #4935 originally added this reset; it was then lost in commit `58daa674e` during the `withMgmtStream` refactor (Sync/Job unification). Compared to before the reset was done AFTER receiveEvents, whereas now it's done before. Now should cover for Sync and Job (which didn't exist in #4935. Hold for long living stream that breaks. Reset prevents two things 1. long living conns going wrong from waiting a long backoff time window before to drive reconn 2. past MaxElapsedTime (3months) long lived conns failures from being treated as "unrecoverable" (hence triggering an full restart of the engine) ## Issue ticket number and link Internal support case (customer agents lost data-plane connectivity during a management maintenance/release window). No public issue. Regression introduced in commit `58daa674e`, which removed the `backOff.Reset()` added by PR #4935. ## Stack <!-- branch-stack --> ### Checklist - [x] Is it a bug fix - [ ] Is a typo/documentation fix - [ ] Is a feature enhancement - [ ] It is a refactor - [ ] Created tests that fail without the change (if possible) > By submitting this pull request, you confirm that you have read and agree to the terms of the [Contributor License Agreement](https://github.com/netbirdio/netbird/blob/main/CONTRIBUTOR_LICENSE_AGREEMENT.md). ## Documentation Select exactly one: - [ ] I added/updated documentation for this change - [x] Documentation is **not needed** for this change (explain why) Internal reconnection-behavior fix in the management gRPC client. No public NetBird CLI, API, or configuration surface changes. ### Docs PR URL (required if "docs added" is checked) Paste the PR link from https://github.com/netbirdio/docs here: N/A <!-- codesmith:footer --> --- <a href="https://app.blacksmith.sh/netbirdio/codesmith/netbird/pr/6883"><picture><source media="(prefers-color-scheme: dark)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"><source media="(prefers-color-scheme: light)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-light-v2.svg"><img alt="View with [code]smith" src="https://pr-comments-assets.blacksmith.sh/codesmith/view-with-codesmith-dark-v2.svg"></picture></a> <a href="https://backend.blacksmith.sh/track/enable-autofix?expires=1787484622&installation_model_id=427504&pr_number=6883&repository=netbirdio%2Fnetbird&return_to=https%3A%2F%2Fgithub.com%2Fnetbirdio%2Fnetbird%2Fpull%2F6883&signature=525b08edbcb94ee6f4ae4226fb1fe8ddb829caa7d4489e8efa639770af9dccdf"><picture><source media="(prefers-color-scheme: dark)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"><source media="(prefers-color-scheme: light)" srcset="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-light.svg"><img alt="Autofix with [code]smith" src="https://pr-comments-assets.blacksmith.sh/codesmith/autofix-with-codesmith-dark.svg"></picture></a> <sup>Need help on this PR? Tag <code>@codesmith-bot</code> with what you need. Autofix is disabled.</sup> <!-- codesmith:autofix:disabled --> <!-- /codesmith:footer --> <!-- This is an auto-generated comment: release notes by coderabbit.ai --> ## Summary by CodeRabbit * **Bug Fixes** * Improved management stream retry behavior. * Retry delays now reset after a stream is successfully established, helping subsequent connection attempts recover more quickly. <!-- end of auto-generated comment: release notes by coderabbit.ai -->
This commit is contained in:
@@ -187,16 +187,16 @@ func (c *GrpcClient) ready() bool {
|
|||||||
// Sync wraps the real client's Sync endpoint call and takes care of retries and encryption/decryption of messages
|
// Sync wraps the real client's Sync endpoint call and takes care of retries and encryption/decryption of messages
|
||||||
// Blocking request. The result will be sent via msgHandler callback function
|
// Blocking request. The result will be sent via msgHandler callback function
|
||||||
func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error {
|
func (c *GrpcClient) Sync(ctx context.Context, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error {
|
||||||
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key) error {
|
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error {
|
||||||
return c.handleSyncStream(ctx, serverPubKey, sysInfo, msgHandler)
|
return c.handleSyncStream(ctx, serverPubKey, sysInfo, msgHandler, backOff)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// Job wraps the real client's Job endpoint call and takes care of retries and encryption/decryption of messages
|
// Job wraps the real client's Job endpoint call and takes care of retries and encryption/decryption of messages
|
||||||
// Blocking request. The result will be sent via msgHandler callback function
|
// Blocking request. The result will be sent via msgHandler callback function
|
||||||
func (c *GrpcClient) Job(ctx context.Context, msgHandler func(msg *proto.JobRequest) *proto.JobResponse) error {
|
func (c *GrpcClient) Job(ctx context.Context, msgHandler func(msg *proto.JobRequest) *proto.JobResponse) error {
|
||||||
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key) error {
|
return c.withMgmtStream(ctx, func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error {
|
||||||
return c.handleJobStream(ctx, serverPubKey, msgHandler)
|
return c.handleJobStream(ctx, serverPubKey, msgHandler, backOff)
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -204,7 +204,7 @@ func (c *GrpcClient) Job(ctx context.Context, msgHandler func(msg *proto.JobRequ
|
|||||||
// It takes care of retries, connection readiness, and fetching server public key.
|
// It takes care of retries, connection readiness, and fetching server public key.
|
||||||
func (c *GrpcClient) withMgmtStream(
|
func (c *GrpcClient) withMgmtStream(
|
||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
handler func(ctx context.Context, serverPubKey wgtypes.Key) error,
|
handler func(ctx context.Context, serverPubKey wgtypes.Key, backOff backoff.BackOff) error,
|
||||||
) error {
|
) error {
|
||||||
backOff := defaultBackoff(ctx)
|
backOff := defaultBackoff(ctx)
|
||||||
operation := func() error {
|
operation := func() error {
|
||||||
@@ -224,7 +224,7 @@ func (c *GrpcClient) withMgmtStream(
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
return handler(ctx, *serverPubKey)
|
return handler(ctx, *serverPubKey, backOff)
|
||||||
}
|
}
|
||||||
|
|
||||||
err := backoff.Retry(operation, backOff)
|
err := backoff.Retry(operation, backOff)
|
||||||
@@ -239,6 +239,7 @@ func (c *GrpcClient) handleJobStream(
|
|||||||
ctx context.Context,
|
ctx context.Context,
|
||||||
serverPubKey wgtypes.Key,
|
serverPubKey wgtypes.Key,
|
||||||
msgHandler func(msg *proto.JobRequest) *proto.JobResponse,
|
msgHandler func(msg *proto.JobRequest) *proto.JobResponse,
|
||||||
|
backOff backoff.BackOff,
|
||||||
) error {
|
) error {
|
||||||
ctx, cancelStream := context.WithCancel(ctx)
|
ctx, cancelStream := context.WithCancel(ctx)
|
||||||
defer cancelStream()
|
defer cancelStream()
|
||||||
@@ -256,6 +257,19 @@ func (c *GrpcClient) handleJobStream(
|
|||||||
|
|
||||||
log.Debug("job stream handshake sent successfully")
|
log.Debug("job stream handshake sent successfully")
|
||||||
|
|
||||||
|
// The stream is up, so reset the backoff. This matters for two reasons,
|
||||||
|
// both caused by the backoff lib not resetting its state on a successful
|
||||||
|
// connection:
|
||||||
|
// 1. Without a reset, after a connect followed by an error the next retry
|
||||||
|
// starts from the accumulated (large) interval instead of retrying
|
||||||
|
// promptly, delaying reconnection.
|
||||||
|
// 2. Worse, once the accumulated elapsed time exceeds MaxElapsedTime, the
|
||||||
|
// next stream error makes NextBackOff() return Stop, so the retry loop
|
||||||
|
// exits immediately. That error is then mislabeled unrecoverable and
|
||||||
|
// bubbles up to trigger a full engine restart / data-plane teardown
|
||||||
|
// instead of a silent reconnection.
|
||||||
|
backOff.Reset()
|
||||||
|
|
||||||
// Main loop: receive, process, respond
|
// Main loop: receive, process, respond
|
||||||
for {
|
for {
|
||||||
jobReq, err := c.receiveJobRequest(ctx, stream, serverPubKey)
|
jobReq, err := c.receiveJobRequest(ctx, stream, serverPubKey)
|
||||||
@@ -371,7 +385,7 @@ func (c *GrpcClient) sendJobResponse(
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error) error {
|
func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes.Key, sysInfo *system.Info, msgHandler func(msg *proto.SyncResponse) error, backOff backoff.BackOff) error {
|
||||||
ctx, cancelStream := context.WithCancel(ctx)
|
ctx, cancelStream := context.WithCancel(ctx)
|
||||||
defer cancelStream()
|
defer cancelStream()
|
||||||
|
|
||||||
@@ -390,6 +404,19 @@ func (c *GrpcClient) handleSyncStream(ctx context.Context, serverPubKey wgtypes.
|
|||||||
c.notifyConnected()
|
c.notifyConnected()
|
||||||
c.setSyncStreamConnected()
|
c.setSyncStreamConnected()
|
||||||
|
|
||||||
|
// The stream is up, so reset the backoff. This matters for two reasons,
|
||||||
|
// both caused by the backoff lib not resetting its state on a successful
|
||||||
|
// connection:
|
||||||
|
// 1. Without a reset, after a connect followed by an error the next retry
|
||||||
|
// starts from the accumulated (large) interval instead of retrying
|
||||||
|
// promptly, delaying reconnection.
|
||||||
|
// 2. Worse, once the accumulated elapsed time exceeds MaxElapsedTime, the
|
||||||
|
// next stream error makes NextBackOff() return Stop, so the retry loop
|
||||||
|
// exits immediately. That error is then mislabeled unrecoverable and
|
||||||
|
// bubbles up to trigger a full engine restart / data-plane teardown
|
||||||
|
// instead of a silent reconnection.
|
||||||
|
backOff.Reset()
|
||||||
|
|
||||||
// blocking until error
|
// blocking until error
|
||||||
err = c.receiveUpdatesEvents(stream, serverPubKey, msgHandler)
|
err = c.receiveUpdatesEvents(stream, serverPubKey, msgHandler)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user