From 90e25b4897a0eb31142818d842ee5309fe9a046a Mon Sep 17 00:00:00 2001 From: bernie-g Date: Tue, 18 Aug 2026 11:37:56 -0400 Subject: [PATCH 1/7] feat(gateway): report active channel count for pool load balancing Counts channels in handleIncomingChannel, the single point every inbound channel passes through whatever opened it, and reports the total every 10s so the platform can route new work to the least busy member of a pool. Kept off the heartbeat deliberately: a heartbeat makes the platform dial back through the relay and write to its database, which is far too costly at the cadence selection needs. A missed report only costs accuracy, so failures are logged at debug and the platform falls back to its own view. --- packages/api/api.go | 19 +++++++++++++++ packages/api/model.go | 4 ++++ packages/gateway-v2/gateway.go | 43 ++++++++++++++++++++++++++++++++++ 3 files changed, 66 insertions(+) diff --git a/packages/api/api.go b/packages/api/api.go index 9bce7e93..905288f2 100644 --- a/packages/api/api.go +++ b/packages/api/api.go @@ -45,6 +45,7 @@ const ( operationCallExchangeRelayCertV1 = "CallExchangeRelayCertV1" operationCallGatewayHeartBeatV1 = "CallGatewayHeartBeatV1" operationCallGatewayHeartBeatV2 = "CallGatewayHeartBeatV2" + operationCallGatewayLoadReportV2 = "CallGatewayLoadReportV2" operationCallBootstrapInstance = "CallBootstrapInstance" operationCallRegisterInstanceRelay = "CallRegisterInstanceRelay" operationCallRegisterOrgRelay = "CallRegisterOrgRelay" @@ -847,6 +848,24 @@ func CallGatewayHeartBeatV2(httpClient *resty.Client, request GatewayHeartbeatRe return nil } +func CallGatewayLoadReportV2(httpClient *resty.Client, request GatewayLoadReportRequest) error { + response, err := httpClient. + R(). + SetHeader("User-Agent", USER_AGENT). + SetBody(request). + Post(fmt.Sprintf("%v/v2/gateways/load", config.INFISICAL_URL)) + + if err != nil { + return NewGenericRequestError(operationCallGatewayLoadReportV2, err) + } + + if response.IsError() { + return NewAPIErrorWithResponse(operationCallGatewayLoadReportV2, response, nil) + } + + return nil +} + func CallOrgRelayHeartBeat(httpClient *resty.Client, request RelayHeartbeatRequest) error { response, err := httpClient. R(). diff --git a/packages/api/model.go b/packages/api/model.go index ec9aeddb..8a5834c1 100644 --- a/packages/api/model.go +++ b/packages/api/model.go @@ -1084,6 +1084,10 @@ type GatewayHeartbeatRequest struct { Capabilities map[string]any `json:"capabilities,omitempty"` } +type GatewayLoadReportRequest struct { + ActiveChannels int64 `json:"activeChannels"` +} + type RelayLoginRequest struct { Method string `json:"method"` Token string `json:"token,omitempty"` diff --git a/packages/gateway-v2/gateway.go b/packages/gateway-v2/gateway.go index 9fdae907..7678d8dd 100644 --- a/packages/gateway-v2/gateway.go +++ b/packages/gateway-v2/gateway.go @@ -57,6 +57,10 @@ const ( const heartbeatInterval = 3 * time.Minute +// Load reports are separate from the heartbeat: a heartbeat makes the platform dial back through the +// relay and write to its database, which is far too costly to run at the cadence pool selection needs. +const loadReportInterval = 10 * time.Second + const GATEWAY_ROUTING_INFO_OID = "1.3.6.1.4.1.12345.100.1" const GATEWAY_ACTOR_OID = "1.3.6.1.4.1.12345.100.2" const PAM_INFO_OID = "1.3.6.1.4.1.12345.100.3" @@ -144,6 +148,11 @@ type Gateway struct { mongoProxies map[string]*mongoProxyEntry mongoProxiesMu sync.Mutex pkcs11Module Pkcs11Module + + // Every channel reaching this gateway is accepted in handleIncomingChannel, whoever opened it, + // so counting there is the only place that cannot be bypassed by a new caller. The platform uses + // it to pick the least loaded member of a gateway pool. + activeChannels atomic.Int64 } // mongoProxyEntry holds a session-level MongoDB proxy with a ready signal. @@ -384,6 +393,36 @@ func (g *Gateway) reapIdleSessions() { } } +// reportLoad publishes the gateway's active channel count so the platform can route new work to the +// least loaded member of a pool. Failures are not surfaced: a missed report only costs accuracy, and +// the platform falls back to its own view when a gateway stops reporting. +func (g *Gateway) reportLoad(ctx context.Context) { + go func() { + ticker := time.NewTicker(loadReportInterval) + defer ticker.Stop() + + var last int64 = -1 + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + count := g.activeChannels.Load() + // Still republish an unchanged count so the platform can tell a quiet gateway from + // one that has stopped reporting. + if err := api.CallGatewayLoadReportV2(g.httpClient, api.GatewayLoadReportRequest{ActiveChannels: count}); err != nil { + log.Debug().Msgf("Load report failed: %v", err) + continue + } + if count != last { + log.Debug().Msgf("Reported %d active channels", count) + last = count + } + } + } + }() +} + func (g *Gateway) registerHeartBeat(ctx context.Context, errCh chan error) { sendHeartbeat := func() error { capabilities := map[string]any{} @@ -575,6 +614,7 @@ func (g *Gateway) startHeartbeatOnce(ctx context.Context, errCh chan error) { defer g.heartbeatMu.Unlock() if !g.heartbeatStarted { g.registerHeartBeat(ctx, errCh) + g.reportLoad(ctx) g.heartbeatStarted = true } } @@ -884,6 +924,9 @@ func (g *Gateway) handleIncomingChannel(newChannel ssh.NewChannel) { } defer channel.Close() + g.activeChannels.Add(1) + defer g.activeChannels.Add(-1) + go ssh.DiscardRequests(requests) // Create mTLS server configuration From 6210b36f8e3ebb353b97d49c940782ee5260f4b5 Mon Sep 17 00:00:00 2001 From: bernie-g Date: Tue, 18 Aug 2026 14:31:44 -0400 Subject: [PATCH 2/7] fix(gateway): bound the load report request The report ran with no deadline, so an endpoint that accepts the connection and then stalls would block the reporting loop indefinitely: no further reports, and no reaction to cancellation either, so shutdown would hang behind it. Each report now runs on a context with a 5s deadline derived from the gateway's own context, which keeps a stall from reaching the next tick and lets cancellation abort a request already in flight. --- packages/api/api.go | 4 +++- packages/gateway-v2/gateway.go | 18 +++++++++++++++--- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/packages/api/api.go b/packages/api/api.go index 905288f2..4174d942 100644 --- a/packages/api/api.go +++ b/packages/api/api.go @@ -1,6 +1,7 @@ package api import ( + "context" "encoding/base64" "errors" "fmt" @@ -848,9 +849,10 @@ func CallGatewayHeartBeatV2(httpClient *resty.Client, request GatewayHeartbeatRe return nil } -func CallGatewayLoadReportV2(httpClient *resty.Client, request GatewayLoadReportRequest) error { +func CallGatewayLoadReportV2(ctx context.Context, httpClient *resty.Client, request GatewayLoadReportRequest) error { response, err := httpClient. R(). + SetContext(ctx). SetHeader("User-Agent", USER_AGENT). SetBody(request). Post(fmt.Sprintf("%v/v2/gateways/load", config.INFISICAL_URL)) diff --git a/packages/gateway-v2/gateway.go b/packages/gateway-v2/gateway.go index 7678d8dd..247459e7 100644 --- a/packages/gateway-v2/gateway.go +++ b/packages/gateway-v2/gateway.go @@ -57,10 +57,16 @@ const ( const heartbeatInterval = 3 * time.Minute -// Load reports are separate from the heartbeat: a heartbeat makes the platform dial back through the -// relay and write to its database, which is far too costly to run at the cadence pool selection needs. +// Reported over the same direct HTTP client as the heartbeat. The relay connection cannot carry this: +// the gateway only accepts channels on it, so it runs platform-to-gateway and has no reverse path. +// Kept off the heartbeat itself because that handler probes back through the relay and writes to the +// platform's database, which is far too costly at the cadence pool selection needs. const loadReportInterval = 10 * time.Second +// Must stay below the interval so a stalled endpoint cannot hold the reporting loop past the next +// tick. Without a bound the loop would also stop noticing cancellation and shutdown would hang. +const loadReportTimeout = 5 * time.Second + const GATEWAY_ROUTING_INFO_OID = "1.3.6.1.4.1.12345.100.1" const GATEWAY_ACTOR_OID = "1.3.6.1.4.1.12345.100.2" const PAM_INFO_OID = "1.3.6.1.4.1.12345.100.3" @@ -396,6 +402,12 @@ func (g *Gateway) reapIdleSessions() { // reportLoad publishes the gateway's active channel count so the platform can route new work to the // least loaded member of a pool. Failures are not surfaced: a missed report only costs accuracy, and // the platform falls back to its own view when a gateway stops reporting. +func (g *Gateway) sendLoadReport(ctx context.Context, count int64) error { + reqCtx, cancel := context.WithTimeout(ctx, loadReportTimeout) + defer cancel() + return api.CallGatewayLoadReportV2(reqCtx, g.httpClient, api.GatewayLoadReportRequest{ActiveChannels: count}) +} + func (g *Gateway) reportLoad(ctx context.Context) { go func() { ticker := time.NewTicker(loadReportInterval) @@ -410,7 +422,7 @@ func (g *Gateway) reportLoad(ctx context.Context) { count := g.activeChannels.Load() // Still republish an unchanged count so the platform can tell a quiet gateway from // one that has stopped reporting. - if err := api.CallGatewayLoadReportV2(g.httpClient, api.GatewayLoadReportRequest{ActiveChannels: count}); err != nil { + if err := g.sendLoadReport(ctx, count); err != nil { log.Debug().Msgf("Load report failed: %v", err) continue } From 1d66d5648dceeffa624db8c0a502f9973e50728e Mon Sep 17 00:00:00 2001 From: bernie-g Date: Thu, 20 Aug 2026 16:10:20 -0400 Subject: [PATCH 3/7] refactor(gateway): rename the load report endpoint to /metrics --- packages/api/api.go | 10 +++++----- packages/api/model.go | 2 +- packages/gateway-v2/gateway.go | 18 +++++++++--------- 3 files changed, 15 insertions(+), 15 deletions(-) diff --git a/packages/api/api.go b/packages/api/api.go index 4174d942..1707cfd2 100644 --- a/packages/api/api.go +++ b/packages/api/api.go @@ -46,7 +46,7 @@ const ( operationCallExchangeRelayCertV1 = "CallExchangeRelayCertV1" operationCallGatewayHeartBeatV1 = "CallGatewayHeartBeatV1" operationCallGatewayHeartBeatV2 = "CallGatewayHeartBeatV2" - operationCallGatewayLoadReportV2 = "CallGatewayLoadReportV2" + operationCallGatewayMetricsReportV2 = "CallGatewayMetricsReportV2" operationCallBootstrapInstance = "CallBootstrapInstance" operationCallRegisterInstanceRelay = "CallRegisterInstanceRelay" operationCallRegisterOrgRelay = "CallRegisterOrgRelay" @@ -849,20 +849,20 @@ func CallGatewayHeartBeatV2(httpClient *resty.Client, request GatewayHeartbeatRe return nil } -func CallGatewayLoadReportV2(ctx context.Context, httpClient *resty.Client, request GatewayLoadReportRequest) error { +func CallGatewayMetricsReportV2(ctx context.Context, httpClient *resty.Client, request GatewayMetricsReportRequest) error { response, err := httpClient. R(). SetContext(ctx). SetHeader("User-Agent", USER_AGENT). SetBody(request). - Post(fmt.Sprintf("%v/v2/gateways/load", config.INFISICAL_URL)) + Post(fmt.Sprintf("%v/v2/gateways/metrics", config.INFISICAL_URL)) if err != nil { - return NewGenericRequestError(operationCallGatewayLoadReportV2, err) + return NewGenericRequestError(operationCallGatewayMetricsReportV2, err) } if response.IsError() { - return NewAPIErrorWithResponse(operationCallGatewayLoadReportV2, response, nil) + return NewAPIErrorWithResponse(operationCallGatewayMetricsReportV2, response, nil) } return nil diff --git a/packages/api/model.go b/packages/api/model.go index 8a5834c1..3188f73c 100644 --- a/packages/api/model.go +++ b/packages/api/model.go @@ -1084,7 +1084,7 @@ type GatewayHeartbeatRequest struct { Capabilities map[string]any `json:"capabilities,omitempty"` } -type GatewayLoadReportRequest struct { +type GatewayMetricsReportRequest struct { ActiveChannels int64 `json:"activeChannels"` } diff --git a/packages/gateway-v2/gateway.go b/packages/gateway-v2/gateway.go index 247459e7..8eadba7f 100644 --- a/packages/gateway-v2/gateway.go +++ b/packages/gateway-v2/gateway.go @@ -61,11 +61,11 @@ const heartbeatInterval = 3 * time.Minute // the gateway only accepts channels on it, so it runs platform-to-gateway and has no reverse path. // Kept off the heartbeat itself because that handler probes back through the relay and writes to the // platform's database, which is far too costly at the cadence pool selection needs. -const loadReportInterval = 10 * time.Second +const metricsReportInterval = 10 * time.Second // Must stay below the interval so a stalled endpoint cannot hold the reporting loop past the next // tick. Without a bound the loop would also stop noticing cancellation and shutdown would hang. -const loadReportTimeout = 5 * time.Second +const metricsReportTimeout = 5 * time.Second const GATEWAY_ROUTING_INFO_OID = "1.3.6.1.4.1.12345.100.1" const GATEWAY_ACTOR_OID = "1.3.6.1.4.1.12345.100.2" @@ -402,15 +402,15 @@ func (g *Gateway) reapIdleSessions() { // reportLoad publishes the gateway's active channel count so the platform can route new work to the // least loaded member of a pool. Failures are not surfaced: a missed report only costs accuracy, and // the platform falls back to its own view when a gateway stops reporting. -func (g *Gateway) sendLoadReport(ctx context.Context, count int64) error { - reqCtx, cancel := context.WithTimeout(ctx, loadReportTimeout) +func (g *Gateway) sendMetricsReport(ctx context.Context, count int64) error { + reqCtx, cancel := context.WithTimeout(ctx, metricsReportTimeout) defer cancel() - return api.CallGatewayLoadReportV2(reqCtx, g.httpClient, api.GatewayLoadReportRequest{ActiveChannels: count}) + return api.CallGatewayMetricsReportV2(reqCtx, g.httpClient, api.GatewayMetricsReportRequest{ActiveChannels: count}) } -func (g *Gateway) reportLoad(ctx context.Context) { +func (g *Gateway) reportMetrics(ctx context.Context) { go func() { - ticker := time.NewTicker(loadReportInterval) + ticker := time.NewTicker(metricsReportInterval) defer ticker.Stop() var last int64 = -1 @@ -422,7 +422,7 @@ func (g *Gateway) reportLoad(ctx context.Context) { count := g.activeChannels.Load() // Still republish an unchanged count so the platform can tell a quiet gateway from // one that has stopped reporting. - if err := g.sendLoadReport(ctx, count); err != nil { + if err := g.sendMetricsReport(ctx, count); err != nil { log.Debug().Msgf("Load report failed: %v", err) continue } @@ -626,7 +626,7 @@ func (g *Gateway) startHeartbeatOnce(ctx context.Context, errCh chan error) { defer g.heartbeatMu.Unlock() if !g.heartbeatStarted { g.registerHeartBeat(ctx, errCh) - g.reportLoad(ctx) + g.reportMetrics(ctx) g.heartbeatStarted = true } } From f5b073ad269b1ec280fe825525e8849b952f5cb4 Mon Sep 17 00:00:00 2001 From: bernie-g Date: Thu, 20 Aug 2026 16:42:02 -0400 Subject: [PATCH 4/7] chore(deps): bump pion/stun to v3.1.5 Clears GO-2026-6163, a panic on a malformed XOR-MAPPED-ADDRESS attribute. Reached through pion/turn in the gateway relay path. v3.1.5 is the first fixed release; later patches pull in dtls and transport bumps this does not need. --- go.mod | 4 ++-- go.sum | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/go.mod b/go.mod index e5acef0f..4dcfc337 100644 --- a/go.mod +++ b/go.mod @@ -181,9 +181,9 @@ require ( github.com/pingcap/log v1.1.1-0.20241212030209-7e3ff8601a2a // indirect github.com/pingcap/tidb/pkg/parser v0.0.0-20250421232622-526b2c79173d // indirect github.com/pion/randutil v0.1.0 // indirect - github.com/pion/stun/v3 v3.0.0 // indirect + github.com/pion/stun/v3 v3.1.5 // indirect github.com/pion/transport/v3 v3.0.7 // indirect - github.com/pion/transport/v4 v4.0.1 // indirect + github.com/pion/transport/v4 v4.0.2 // indirect github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect github.com/rivo/uniseg v0.2.0 // indirect github.com/shopspring/decimal v1.4.0 // indirect diff --git a/go.sum b/go.sum index 993a0108..c69d3a3a 100644 --- a/go.sum +++ b/go.sum @@ -570,12 +570,12 @@ github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so= github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= -github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw= -github.com/pion/stun/v3 v3.0.0/go.mod h1:HvCN8txt8mwi4FBvS3EmDghW6aQJ24T+y+1TKjB5jyU= +github.com/pion/stun/v3 v3.1.5 h1:Y1FHlhaI6+4UoC5i/zQf4F7JvdZtB24/05oyy/GF1x8= +github.com/pion/stun/v3 v3.1.5/go.mod h1:zRUghXSQU32Lx5orJsz3uYMkIihweXb3mu5gIns02fs= github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0= github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo= -github.com/pion/transport/v4 v4.0.1 h1:sdROELU6BZ63Ab7FrOLn13M6YdJLY20wldXW2Cu2k8o= -github.com/pion/transport/v4 v4.0.1/go.mod h1:nEuEA4AD5lPdcIegQDpVLgNoDGreqM/YqmEx3ovP4jM= +github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6Tk= +github.com/pion/transport/v4 v4.0.2/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= From 5e9ed357eab52d2eb317833d3137d28d396188a8 Mon Sep 17 00:00:00 2001 From: bernie-g Date: Thu, 20 Aug 2026 16:57:47 -0400 Subject: [PATCH 5/7] chore(deps): sync the e2e module with the stun bump The e2e module replaces the root module, so its go.sum needs the same versions or every e2e job fails with "updates to go.mod needed". --- e2e/go.mod | 4 ++-- e2e/go.sum | 8 ++++---- 2 files changed, 6 insertions(+), 6 deletions(-) diff --git a/e2e/go.mod b/e2e/go.mod index ed358a56..0fc80dc2 100644 --- a/e2e/go.mod +++ b/e2e/go.mod @@ -250,9 +250,9 @@ require ( github.com/pion/dtls/v3 v3.1.4 // indirect github.com/pion/logging v0.2.4 // indirect github.com/pion/randutil v0.1.0 // indirect - github.com/pion/stun/v3 v3.0.0 // indirect + github.com/pion/stun/v3 v3.1.5 // indirect github.com/pion/transport/v3 v3.0.7 // indirect - github.com/pion/transport/v4 v4.0.1 // indirect + github.com/pion/transport/v4 v4.0.2 // indirect github.com/pion/turn/v4 v4.0.0 // indirect github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect github.com/pkg/errors v0.9.1 // indirect diff --git a/e2e/go.sum b/e2e/go.sum index e873f8fb..9ed8bee5 100644 --- a/e2e/go.sum +++ b/e2e/go.sum @@ -857,12 +857,12 @@ github.com/pion/logging v0.2.4 h1:tTew+7cmQ+Mc1pTBLKH2puKsOvhm32dROumOZ655zB8= github.com/pion/logging v0.2.4/go.mod h1:DffhXTKYdNZU+KtJ5pyQDjvOAh/GsNSyv1lbkFbe3so= github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA= github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8= -github.com/pion/stun/v3 v3.0.0 h1:4h1gwhWLWuZWOJIJR9s2ferRO+W3zA/b6ijOI6mKzUw= -github.com/pion/stun/v3 v3.0.0/go.mod h1:HvCN8txt8mwi4FBvS3EmDghW6aQJ24T+y+1TKjB5jyU= +github.com/pion/stun/v3 v3.1.5 h1:Y1FHlhaI6+4UoC5i/zQf4F7JvdZtB24/05oyy/GF1x8= +github.com/pion/stun/v3 v3.1.5/go.mod h1:zRUghXSQU32Lx5orJsz3uYMkIihweXb3mu5gIns02fs= github.com/pion/transport/v3 v3.0.7 h1:iRbMH05BzSNwhILHoBoAPxoB9xQgOaJk+591KC9P1o0= github.com/pion/transport/v3 v3.0.7/go.mod h1:YleKiTZ4vqNxVwh77Z0zytYi7rXHl7j6uPLGhhz9rwo= -github.com/pion/transport/v4 v4.0.1 h1:sdROELU6BZ63Ab7FrOLn13M6YdJLY20wldXW2Cu2k8o= -github.com/pion/transport/v4 v4.0.1/go.mod h1:nEuEA4AD5lPdcIegQDpVLgNoDGreqM/YqmEx3ovP4jM= +github.com/pion/transport/v4 v4.0.2 h1:ifYlPqNwsy6aKQ9y8yzxXlHae5431ZrH2avkD/Rn6Tk= +github.com/pion/transport/v4 v4.0.2/go.mod h1:06hFI+jCFcok2X2MekVufNZ/uzNZXivGBPfviSVcjgM= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c h1:+mdjkGKdHQG3305AYmdv1U2eRNDiU2ErMBj1gwrq8eQ= github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c/go.mod h1:7rwL4CYBLnjLxUqIJNnCWiEdr3bn6IUYi15bNlnbCCU= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= From d9c4392b1d355a52b38a6b81f2001bd4a126d770 Mon Sep 17 00:00:00 2001 From: bernie-g Date: Mon, 24 Aug 2026 20:56:28 -0400 Subject: [PATCH 6/7] chore(gateway): trim comments and correct the reporting note The doc on the reporting loop said the platform falls back to its own view when a gateway stops reporting. It no longer does: a non-reporting member takes its whole pool off load-aware selection. --- packages/gateway-v2/gateway.go | 21 +++++++-------------- 1 file changed, 7 insertions(+), 14 deletions(-) diff --git a/packages/gateway-v2/gateway.go b/packages/gateway-v2/gateway.go index 8eadba7f..38f4f589 100644 --- a/packages/gateway-v2/gateway.go +++ b/packages/gateway-v2/gateway.go @@ -57,14 +57,11 @@ const ( const heartbeatInterval = 3 * time.Minute -// Reported over the same direct HTTP client as the heartbeat. The relay connection cannot carry this: -// the gateway only accepts channels on it, so it runs platform-to-gateway and has no reverse path. -// Kept off the heartbeat itself because that handler probes back through the relay and writes to the -// platform's database, which is far too costly at the cadence pool selection needs. +// Kept off the heartbeat because that handler probes back through the relay and writes to the +// platform's database, which is far too costly at this cadence. const metricsReportInterval = 10 * time.Second -// Must stay below the interval so a stalled endpoint cannot hold the reporting loop past the next -// tick. Without a bound the loop would also stop noticing cancellation and shutdown would hang. +// Below the interval, so a stalled endpoint cannot hold the loop past the next tick or block shutdown. const metricsReportTimeout = 5 * time.Second const GATEWAY_ROUTING_INFO_OID = "1.3.6.1.4.1.12345.100.1" @@ -155,9 +152,7 @@ type Gateway struct { mongoProxiesMu sync.Mutex pkcs11Module Pkcs11Module - // Every channel reaching this gateway is accepted in handleIncomingChannel, whoever opened it, - // so counting there is the only place that cannot be bypassed by a new caller. The platform uses - // it to pick the least loaded member of a gateway pool. + // Counted in handleIncomingChannel, the one place no caller can bypass. activeChannels atomic.Int64 } @@ -399,9 +394,8 @@ func (g *Gateway) reapIdleSessions() { } } -// reportLoad publishes the gateway's active channel count so the platform can route new work to the -// least loaded member of a pool. Failures are not surfaced: a missed report only costs accuracy, and -// the platform falls back to its own view when a gateway stops reporting. +// Publishes the active channel count so the platform can route to the least loaded pool member. +// A gateway that stops reporting takes its whole pool off load-aware selection. func (g *Gateway) sendMetricsReport(ctx context.Context, count int64) error { reqCtx, cancel := context.WithTimeout(ctx, metricsReportTimeout) defer cancel() @@ -420,8 +414,7 @@ func (g *Gateway) reportMetrics(ctx context.Context) { return case <-ticker.C: count := g.activeChannels.Load() - // Still republish an unchanged count so the platform can tell a quiet gateway from - // one that has stopped reporting. + // Republish unchanged, so a quiet gateway is distinguishable from a silent one. if err := g.sendMetricsReport(ctx, count); err != nil { log.Debug().Msgf("Load report failed: %v", err) continue From 37c4648af15d363a56463722bea855a7f3f39d93 Mon Sep 17 00:00:00 2001 From: bernie-g Date: Mon, 24 Aug 2026 21:22:54 -0400 Subject: [PATCH 7/7] fix(gateway): bound the active channel count and back off failed reports A handler that never returns left the count high for the life of the process, so the gateway reported itself permanently busy and stopped attracting work. The count resets when a relay connection is established, since channels do not outlive the connection they arrived on, and the decrement is floored so a handler from the previous connection cannot push it negative. Reports against a platform with no metrics endpoint retried every 10s forever. After five consecutive failures the interval drops to five minutes and warns once, naming the consequence: pools containing the gateway select at random. Backs off rather than stopping, so an upgraded platform recovers without a gateway restart. reportMetrics becomes startMetricsReport, since it spawns the loop. --- packages/gateway-v2/gateway.go | 63 +++++++++++++++------ packages/gateway-v2/release_channel_test.go | 38 +++++++++++++ 2 files changed, 85 insertions(+), 16 deletions(-) create mode 100644 packages/gateway-v2/release_channel_test.go diff --git a/packages/gateway-v2/gateway.go b/packages/gateway-v2/gateway.go index 38f4f589..52974335 100644 --- a/packages/gateway-v2/gateway.go +++ b/packages/gateway-v2/gateway.go @@ -64,6 +64,9 @@ const metricsReportInterval = 10 * time.Second // Below the interval, so a stalled endpoint cannot hold the loop past the next tick or block shutdown. const metricsReportTimeout = 5 * time.Second +const metricsReportFailuresBeforeBackoff = 5 +const metricsReportBackoff = 5 * time.Minute + const GATEWAY_ROUTING_INFO_OID = "1.3.6.1.4.1.12345.100.1" const GATEWAY_ACTOR_OID = "1.3.6.1.4.1.12345.100.2" const PAM_INFO_OID = "1.3.6.1.4.1.12345.100.3" @@ -394,35 +397,47 @@ func (g *Gateway) reapIdleSessions() { } } -// Publishes the active channel count so the platform can route to the least loaded pool member. -// A gateway that stops reporting takes its whole pool off load-aware selection. func (g *Gateway) sendMetricsReport(ctx context.Context, count int64) error { reqCtx, cancel := context.WithTimeout(ctx, metricsReportTimeout) defer cancel() return api.CallGatewayMetricsReportV2(reqCtx, g.httpClient, api.GatewayMetricsReportRequest{ActiveChannels: count}) } -func (g *Gateway) reportMetrics(ctx context.Context) { +// A gateway that stops reporting takes its whole pool off load-aware selection. +func (g *Gateway) startMetricsReport(ctx context.Context) { go func() { - ticker := time.NewTicker(metricsReportInterval) - defer ticker.Stop() + delay := metricsReportInterval + failures := 0 var last int64 = -1 for { select { case <-ctx.Done(): return - case <-ticker.C: - count := g.activeChannels.Load() - // Republish unchanged, so a quiet gateway is distinguishable from a silent one. - if err := g.sendMetricsReport(ctx, count); err != nil { - log.Debug().Msgf("Load report failed: %v", err) - continue + case <-time.After(delay): + } + + count := g.activeChannels.Load() + // Republish unchanged, so a quiet gateway is distinguishable from a silent one. + if err := g.sendMetricsReport(ctx, count); err != nil { + failures++ + if failures == metricsReportFailuresBeforeBackoff { + log.Warn().Err(err).Msgf("Metrics report failing; backing off to %s. Pools containing this gateway will select at random until it succeeds", metricsReportBackoff) } - if count != last { - log.Debug().Msgf("Reported %d active channels", count) - last = count + if failures >= metricsReportFailuresBeforeBackoff { + delay = metricsReportBackoff } + continue + } + + if failures >= metricsReportFailuresBeforeBackoff { + log.Info().Msg("Metrics report recovered") + } + failures = 0 + delay = metricsReportInterval + if count != last { + log.Debug().Msgf("Reported %d active channels", count) + last = count } } }() @@ -619,7 +634,7 @@ func (g *Gateway) startHeartbeatOnce(ctx context.Context, errCh chan error) { defer g.heartbeatMu.Unlock() if !g.heartbeatStarted { g.registerHeartBeat(ctx, errCh) - g.reportMetrics(ctx) + g.startMetricsReport(ctx) g.heartbeatStarted = true } } @@ -695,6 +710,9 @@ func (g *Gateway) handleConnection(client *ssh.Client) error { client.Close() }() + // Channels do not outlive their connection, so anything still counted is a handler that hung. + g.activeChannels.Store(0) + // Handle incoming channels from the server channels := client.HandleChannelOpen("direct-tcpip") if channels == nil { @@ -921,6 +939,19 @@ func (g *Gateway) validateHostCertificate(cert *ssh.Certificate, hostname string return nil } +// Floored: a handler from a previous connection must not decrement past the reset. +func (g *Gateway) releaseChannel() { + for { + current := g.activeChannels.Load() + if current <= 0 { + return + } + if g.activeChannels.CompareAndSwap(current, current-1) { + return + } + } +} + func (g *Gateway) handleIncomingChannel(newChannel ssh.NewChannel) { channel, requests, err := newChannel.Accept() if err != nil { @@ -930,7 +961,7 @@ func (g *Gateway) handleIncomingChannel(newChannel ssh.NewChannel) { defer channel.Close() g.activeChannels.Add(1) - defer g.activeChannels.Add(-1) + defer g.releaseChannel() go ssh.DiscardRequests(requests) diff --git a/packages/gateway-v2/release_channel_test.go b/packages/gateway-v2/release_channel_test.go new file mode 100644 index 00000000..2c35a41f --- /dev/null +++ b/packages/gateway-v2/release_channel_test.go @@ -0,0 +1,38 @@ +package gatewayv2 + +import ( + "sync" + "testing" +) + +func TestReleaseChannelNeverGoesNegative(t *testing.T) { + g := &Gateway{} + g.activeChannels.Add(1) + g.activeChannels.Add(1) + + // Reset with handlers still in flight, as on a relay reconnect. + g.activeChannels.Store(0) + + var wg sync.WaitGroup + for i := 0; i < 2; i++ { + wg.Add(1) + go func() { + defer wg.Done() + g.releaseChannel() + }() + } + wg.Wait() + + if got := g.activeChannels.Load(); got != 0 { + t.Fatalf("count went to %d, want 0", got) + } +} + +func TestReleaseChannelDecrements(t *testing.T) { + g := &Gateway{} + g.activeChannels.Add(3) + g.releaseChannel() + if got := g.activeChannels.Load(); got != 2 { + t.Fatalf("count is %d, want 2", got) + } +}