diff --git a/management/internals/server/server.go b/management/internals/server/server.go index c0f0db012..b4183afc9 100644 --- a/management/internals/server/server.go +++ b/management/internals/server/server.go @@ -470,6 +470,11 @@ func preferHTTP1ForDualProtoClients(base *tls.Config) *tls.Config { // serveMultiplexed splits the shared listener by protocol: HTTP/2 connections // go to the native gRPC transport (see preferHTTP1ForDualProtoClients for why // they are all gRPC), everything else is served by net/http. +// +// Content-type based classification cannot be used here: cmux's SendSettings +// matchers greet non-matching HTTP/2 connections and corrupt them for any +// subsequent handler, while read-only matchers deadlock grpc-go clients, +// which do not send HEADERS until they receive the server SETTINGS frame. func (s *BaseServer) serveMultiplexed(ctx context.Context, listener net.Listener, grpcServer *grpc.Server, handler http.Handler, tlsEnabled bool) { mux := cmux.New(listener) grpcListener := mux.Match(cmux.HTTP2()) diff --git a/management/internals/shared/grpc/token_mgr.go b/management/internals/shared/grpc/token_mgr.go index 14d9088e5..1d7775d86 100644 --- a/management/internals/shared/grpc/token_mgr.go +++ b/management/internals/shared/grpc/token_mgr.go @@ -165,10 +165,11 @@ func (m *TimeBasedAuthSecretsManager) runRefreshJob(job *refreshJob) { } func refreshInterval(ttl time.Duration) time.Duration { - if ttl <= 0 { - ttl = defaultDuration + interval := ttl / 4 * 3 + if interval <= 0 { + interval = defaultDuration / 4 * 3 } - return ttl / 4 * 3 + return interval } // SetupRefresh starts peer credentials refresh diff --git a/management/internals/shared/grpc/token_refresh_scheduler.go b/management/internals/shared/grpc/token_refresh_scheduler.go index dbf5d73b1..34921b12a 100644 --- a/management/internals/shared/grpc/token_refresh_scheduler.go +++ b/management/internals/shared/grpc/token_refresh_scheduler.go @@ -4,6 +4,7 @@ import ( "container/heap" "context" "sync" + "sync/atomic" "time" ) @@ -27,6 +28,7 @@ type refreshJob struct { interval time.Duration nextRun time.Time index int + cancelled atomic.Bool } type refreshJobHeap []*refreshJob @@ -95,6 +97,7 @@ func (s *refreshScheduler) schedule(job *refreshJob) { func (s *refreshScheduler) cancel(job *refreshJob) { s.mu.Lock() defer s.mu.Unlock() + job.cancelled.Store(true) if job.index >= 0 { heap.Remove(&s.jobs, job.index) } @@ -147,6 +150,9 @@ func (s *refreshScheduler) loop() { func (s *refreshScheduler) worker() { for job := range s.work { + if job.cancelled.Load() { + continue + } s.run(job) } } diff --git a/management/internals/shared/grpc/token_refresh_scheduler_test.go b/management/internals/shared/grpc/token_refresh_scheduler_test.go new file mode 100644 index 000000000..c5c284f6b --- /dev/null +++ b/management/internals/shared/grpc/token_refresh_scheduler_test.go @@ -0,0 +1,56 @@ +package grpc + +import ( + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestRefreshInterval(t *testing.T) { + defaultInterval := defaultDuration / 4 * 3 + + require.Equal(t, 9*time.Hour, refreshInterval(12*time.Hour)) + require.Equal(t, defaultInterval, refreshInterval(0)) + require.Equal(t, defaultInterval, refreshInterval(-time.Second)) + require.Equal(t, defaultInterval, refreshInterval(3*time.Nanosecond)) + require.Positive(t, refreshInterval(4*time.Nanosecond)) +} + +func TestWorkerSkipsCancelledJob(t *testing.T) { + var ran atomic.Int32 + scheduler := newRefreshScheduler(func(*refreshJob) { + ran.Add(1) + }) + + cancelledJob := &refreshJob{interval: time.Hour} + cancelledJob.cancelled.Store(true) + liveJob := &refreshJob{interval: time.Hour} + + scheduler.work <- cancelledJob + scheduler.work <- liveJob + + require.Eventually(t, func() bool { + return ran.Load() == 1 + }, 2*time.Second, 10*time.Millisecond, "live job should run exactly once, cancelled job never") +} + +func TestCancelBeforeFirePreventsRun(t *testing.T) { + var ran atomic.Int32 + scheduler := newRefreshScheduler(func(*refreshJob) { + ran.Add(1) + }) + + job := &refreshJob{interval: 50 * time.Millisecond} + scheduler.schedule(job) + scheduler.cancel(job) + + time.Sleep(150 * time.Millisecond) + require.Zero(t, ran.Load(), "cancelled job must never fire") + + scheduler.mu.Lock() + heapLen := len(scheduler.jobs) + scheduler.mu.Unlock() + require.Zero(t, heapLen) +}