From 74c57d59943e5d144b2de5d0e874da5fa4a9a74d Mon Sep 17 00:00:00 2001 From: "C, Amarnath" Date: Thu, 1 Oct 2026 21:32:18 +0530 Subject: [PATCH] fix: stop one slow device from blocking the others (#1247) Move the device auth wait off the request worker to the caller's goroutine so one unauthenticated device no longer blocks every other device's setup. --- internal/usecase/devices/wsman/message.go | 145 ++++++++++++++---- .../usecase/devices/wsman/message_test.go | 104 +++++++++++++ 2 files changed, 221 insertions(+), 28 deletions(-) diff --git a/internal/usecase/devices/wsman/message.go b/internal/usecase/devices/wsman/message.go index ad0f4ddc2..b02a15d81 100644 --- a/internal/usecase/devices/wsman/message.go +++ b/internal/usecase/devices/wsman/message.go @@ -4,6 +4,7 @@ import ( "context" gotls "crypto/tls" "errors" + "fmt" "net" "sync" "time" @@ -69,6 +70,11 @@ import ( const ( deviceCallBuffer = 100 maxReadRecords = 390 + + // The queued closure is sub-millisecond now that the authentication wait + // runs on the caller's goroutine, so logging every one of them would bury + // the case worth seeing: the Worker being held after all. + workerStallThreshold = 20 * time.Millisecond ) var ( @@ -133,7 +139,17 @@ func (g GoWSMANMessages) Worker() { for { select { case request := <-requestQueue: + start := time.Now() + request() + + // The closure must never block: anything slow (the authentication + // wait) is handed back to the caller instead. Only report a request + // that held the Worker anyway, or that left work queued behind it. + if held := time.Since(start); held >= workerStallThreshold || len(requestQueue) > 0 { + g.log.Debug(fmt.Sprintf("WSMAN_QUEUE: request held the worker for %s, %d still queued", held, len(requestQueue))) + } + time.Sleep(queueTickTime) case <-shutdownSignal: return @@ -141,8 +157,16 @@ func (g GoWSMANMessages) Worker() { } } +// setupResult carries the connection entry off the Worker goroutine. awaitAuth +// is non-nil when the entry still has to finish authenticating; the caller runs +// it on its own goroutine so the Worker is free for the next device. +type setupResult struct { + entry *ConnectionEntry + awaitAuth func(context.Context) (*ConnectionEntry, error) +} + func (g GoWSMANMessages) SetupWsmanClient(ctx context.Context, device entity.Device, isRedirection, logAMTMessages bool) (Management, error) { - resultChan := make(chan *ConnectionEntry, 1) + resultChan := make(chan setupResult, 1) errChan := make(chan error, 1) // Queue the request requestQueue <- func() { @@ -182,23 +206,35 @@ func (g GoWSMANMessages) SetupWsmanClient(ctx context.Context, device entity.Dev } connection.WsmanMessages = wsman.NewMessages(cp) - resultChan <- connection - } else { - resultChan <- g.setupWsmanClientInternal(device, isRedirection, logAMTMessages) + resultChan <- setupResult{entry: connection} + + return } + + entry, awaitAuth := g.setupWsmanClientInternal(device, isRedirection, logAMTMessages) + resultChan <- setupResult{entry: entry, awaitAuth: awaitAuth} } select { case err := <-errChan: return nil, err case result := <-resultChan: - return result, nil + if result.awaitAuth != nil { + entry, err := result.awaitAuth(ctx) + if err != nil { + return nil, err + } + + return entry, nil + } + + return result.entry, nil case <-ctx.Done(): return nil, ErrCancelled.Wrap("SetupWsmanClient", "ctx.Done", ctx.Err()) } } -func (g GoWSMANMessages) setupWsmanClientInternal(device entity.Device, isRedirection, logAMTMessages bool) *ConnectionEntry { +func (g GoWSMANMessages) setupWsmanClientInternal(device entity.Device, isRedirection, logAMTMessages bool) (conn *ConnectionEntry, awaitAuth func(context.Context) (*ConnectionEntry, error)) { clientParams := client.Parameters{ Target: device.Hostname, Username: device.Username, @@ -226,34 +262,25 @@ func (g GoWSMANMessages) setupWsmanClientInternal(device entity.Device, isRedire RemoveConnection(device.GUID) }) - return entry + return entry, nil } else if entry.IsCIRA { entry.WsmanMessages = wsman.NewMessages(clientParams) - return entry + return entry, nil } - ticker := time.NewTicker(waitForAuthTickTime) - - defer ticker.Stop() - - timeout := time.After(waitForAuth) + // Hand the wait back to the caller instead of running it here. Capture + // the client now, on the Worker goroutine: entry.WsmanMessages is + // assigned without a lock elsewhere, so the caller must not re-read it. + pending := entry.WsmanMessages.Client - for { - select { - case <-ticker.C: - if entry.WsmanMessages.Client.IsAuthenticated() { - return entry - } - case <-timeout: - newEntry := &ConnectionEntry{ - WsmanMessages: wsman.NewMessages(clientParams), - Timer: timer, - } - SetConnectionEntry(device.GUID, newEntry) + // The one line that evidences the handoff: paired with WSMAN_QUEUE + // reporting a sub-millisecond hold, it shows the device is being waited + // on without the Worker being held for it. + g.log.Debug("WSMAN_SETUP: cached connection not authenticated yet, deferring the wait to the caller for guid " + device.GUID) - return newEntry - } + return entry, func(ctx context.Context) (*ConnectionEntry, error) { + return waitForAuthentication(ctx, device.GUID, entry, pending, clientParams, timer) } } @@ -266,7 +293,51 @@ func (g GoWSMANMessages) setupWsmanClientInternal(device entity.Device, isRedire newEntry.WsmanMessages.Client.IsAuthenticated() SetConnectionEntry(device.GUID, newEntry) - return newEntry + return newEntry, nil +} + +// waitForAuthentication polls until an in-flight request finishes the device +// handshake, falling back to a fresh client if it never does. It runs on the +// caller's goroutine rather than inside the queued closure, so a device that is +// slow to authenticate no longer delays every other device's setup (#1247). +func waitForAuthentication(ctx context.Context, guid string, entry *ConnectionEntry, pending client.WSMan, clientParams client.Parameters, timer *time.Timer) (*ConnectionEntry, error) { + ticker := time.NewTicker(waitForAuthTickTime) + + defer ticker.Stop() + + timeout := time.After(waitForAuth) + + for { + select { + case <-ticker.C: + if pending.IsAuthenticated() { + // The cached entry keeps its own expiry timer; ours is surplus. + timer.Stop() + + return entry, nil + } + case <-timeout: + newEntry := &ConnectionEntry{ + WsmanMessages: wsman.NewMessages(clientParams), + Timer: timer, + } + + cached := replaceConnectionEntry(guid, entry, newEntry) + if cached != newEntry { + // Another waiter replaced the entry first; drop ours so its + // expiry timer cannot evict the entry that is now cached. + timer.Stop() + } + + return cached, nil + case <-ctx.Done(): + // Nothing takes ownership of the timer on this path; left armed it + // would evict whatever is cached for the device 90s from now. + timer.Stop() + + return nil, ErrCancelled.Wrap("SetupWsmanClient", "ctx.Done", ctx.Err()) + } + } } // RemoveConnection safely deletes a connection entry from the global map. @@ -293,6 +364,24 @@ func SetConnectionEntry(guid string, entry *ConnectionEntry) { connections[guid] = entry } +// replaceConnectionEntry swaps stale for replacement under the map lock and +// returns the entry that is actually cached. Waiters on the same unauthenticated +// device all time out together; the first one installs its entry and the rest +// get it back, so a burst of requests shares one client instead of each caching +// its own. +func replaceConnectionEntry(guid string, stale, replacement *ConnectionEntry) *ConnectionEntry { + connectionsMu.Lock() + defer connectionsMu.Unlock() + + if current, ok := connections[guid]; ok && current != stale { + return current + } + + connections[guid] = replacement + + return replacement +} + // HasConnections safely checks whether any connections exist. func HasConnections() bool { connectionsMu.RLock() diff --git a/internal/usecase/devices/wsman/message_test.go b/internal/usecase/devices/wsman/message_test.go index 7a90ac67a..5e3c639c0 100644 --- a/internal/usecase/devices/wsman/message_test.go +++ b/internal/usecase/devices/wsman/message_test.go @@ -2,6 +2,7 @@ package wsman import ( "context" + "sync" "testing" "time" @@ -9,7 +10,10 @@ import ( "github.com/stretchr/testify/require" gwmconfig "github.com/device-management-toolkit/go-wsman-messages/v2/pkg/config" + "github.com/device-management-toolkit/go-wsman-messages/v2/pkg/wsman" + "github.com/device-management-toolkit/go-wsman-messages/v2/pkg/wsman/client" + "github.com/device-management-toolkit/console/config" "github.com/device-management-toolkit/console/internal/entity" dto "github.com/device-management-toolkit/console/internal/entity/dto/v1" "github.com/device-management-toolkit/console/pkg/logger" @@ -173,3 +177,103 @@ func TestDestroyWsmanClient_MissingEntryIsNoop(t *testing.T) { // Should not panic when the entry is absent. g.DestroyWsmanClient(dto.Device{GUID: "destroy-missing-entry"}) } + +// TestSetupWsmanClientSlowDeviceDoesNotDelayOthers is a regression test for +// #1247: a device that never authenticates made every other device's setup +// time out, because the wait for authentication ran inside the closure drained +// by the single Worker goroutine. The wait itself is unchanged; it now runs on +// the caller's goroutine, so the Worker stays free. +func TestSetupWsmanClientSlowDeviceDoesNotDelayOthers(t *testing.T) { //nolint:paralleltest // mutates package-level state (requestQueue, waitForAuth, connections) + const ( + slowGUID = "issue-1247-slow-device" + healthyGUID = "issue-1247-healthy-device" + budget = 3 * time.Second + ) + + if config.ConsoleConfig == nil { + config.ConsoleConfig = &config.Config{} + + t.Cleanup(func() { config.ConsoleConfig = nil }) + } + + // Keep the slow device waiting well past the budget without making the + // test sit through the production 30s. + origWait := waitForAuth + waitForAuth = time.Minute + + t.Cleanup(func() { waitForAuth = origWait }) + t.Cleanup(func() { RemoveConnection(slowGUID); RemoveConnection(healthyGUID) }) + + // A cached, unauthenticated entry pointing at a black-hole address: the + // authentication wait will run to its full timeout. + SetConnectionEntry(slowGUID, &ConnectionEntry{ + WsmanMessages: wsman.NewMessages(client.Parameters{ + Target: "192.0.2.1", Username: "u", Password: "p", UseDigest: true, + }), + Timer: time.AfterFunc(time.Hour, func() {}), + }) + require.False(t, GetConnectionEntry(slowGUID).WsmanMessages.Client.IsAuthenticated(), + "the slow device must start out unauthenticated for this test to exercise the wait") + + stopWorker := make(chan struct{}) + workerDone := make(chan struct{}) + + go func() { + defer close(workerDone) + + for { + select { + case <-stopWorker: + return + case req := <-requestQueue: + req() + // Stands in for the Worker's queueTickTime throttle without + // mutating that package-level var out from under it. + time.Sleep(time.Millisecond) + } + } + }() + + t.Cleanup(func() { + close(stopWorker) + + select { + case <-workerDone: + case <-time.After(time.Second): + } + }) + + g := NewGoWSMANMessages(logger.New("error"), passthroughCryptor{}) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var wg sync.WaitGroup + + wg.Add(1) + + go func() { + defer wg.Done() + + _, _ = g.SetupWsmanClient(ctx, entity.Device{GUID: slowGUID, Hostname: "192.0.2.1", Username: "u", Password: "p"}, false, false) + }() + + t.Cleanup(wg.Wait) + + // Give the slow request time to reach the Worker and start waiting. + time.Sleep(50 * time.Millisecond) + + done := make(chan error, 1) + + go func() { + _, err := g.SetupWsmanClient(ctx, entity.Device{GUID: healthyGUID, Hostname: "192.0.2.2", Username: "u", Password: "p"}, false, false) + done <- err + }() + + select { + case err := <-done: + require.NoError(t, err, "the healthy device must set up successfully") + case <-time.After(budget): + t.Fatal("a slow device is still delaying every other device's setup (#1247)") + } +}