mirror of
https://github.com/juanfont/headscale.git
synced 2026-10-28 01:06:50 +09:00
state: stop the auth cache TTL sweeper on Close
The expirable LRU ran its sweeper forever, one goroutine per State. Purge first so pending waiters get ErrRegistrationExpired instead of hanging. The OIDC state cache is closed at server shutdown for the same reason.
This commit is contained in:
1 parent
5f39a68b39
commit
cd62031b98
5 files changed
+128
-35
No files matched your search
@@ -996,6 +996,11 @@ func (h *Headscale) Serve() error {
|
|||||||
info("closing socket listener")
|
info("closing socket listener")
|
||||||
socketListener.Close()
|
socketListener.Close()
|
||||||
|
|
||||||
|
// The OIDC state cache runs its own TTL sweeper goroutine.
|
||||||
|
if p, ok := h.authProvider.(*AuthProviderOIDC); ok {
|
||||||
|
p.authCache.Close()
|
||||||
|
}
|
||||||
|
|
||||||
// Close state connections
|
// Close state connections
|
||||||
info("closing state and database")
|
info("closing state and database")
|
||||||
|
|
||||||
|
|||||||
@@ -229,6 +229,7 @@ func TestGetAuthInfoFromStateSingleUse(t *testing.T) {
|
|||||||
a := &AuthProviderOIDC{
|
a := &AuthProviderOIDC{
|
||||||
authCache: expirable.NewLRU[string, AuthInfo](16, nil, time.Minute),
|
authCache: expirable.NewLRU[string, AuthInfo](16, nil, time.Minute),
|
||||||
}
|
}
|
||||||
|
t.Cleanup(a.authCache.Close)
|
||||||
a.authCache.Add("state-x", AuthInfo{Registration: true})
|
a.authCache.Add("state-x", AuthInfo{Registration: true})
|
||||||
|
|
||||||
got := a.getAuthInfoFromState("state-x")
|
got := a.getAuthInfoFromState("state-x")
|
||||||
@@ -336,6 +337,7 @@ func newOIDCBrowserWithPrefix(t *testing.T, prefix string) *oidcBrowser {
|
|||||||
},
|
},
|
||||||
)
|
)
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
|
t.Cleanup(provider.authCache.Close)
|
||||||
|
|
||||||
app.authProvider = provider
|
app.authProvider = provider
|
||||||
router = app.createRouter(nil, nil)
|
router = app.createRouter(nil, nil)
|
||||||
|
|||||||
@@ -15,19 +15,14 @@ import (
|
|||||||
//
|
//
|
||||||
// TODO(kradalby): These tests wait for real-time grace periods and
|
// TODO(kradalby): These tests wait for real-time grace periods and
|
||||||
// GC intervals (up to 60s). testing/synctest would allow instant
|
// GC intervals (up to 60s). testing/synctest would allow instant
|
||||||
// fake-clock advancement, but three blockers prevent adoption
|
// fake-clock advancement, but two blockers prevent adoption
|
||||||
// as of Go 1.26:
|
// as of Go 1.26:
|
||||||
//
|
//
|
||||||
// 1. golang-lru/v2/expirable janitor goroutine: No Close() method;
|
// 1. database/sql internal goroutines: Uses [sync.RWMutex] which is not
|
||||||
// the deleteExpired ticker goroutine never exits because the done
|
|
||||||
// channel is never closed (documented as a v3 TODO upstream).
|
|
||||||
// - https://github.com/hashicorp/golang-lru/blob/v2.0.7/expirable/expirable_lru.go#L78-L81
|
|
||||||
//
|
|
||||||
// 2. database/sql internal goroutines: Uses [sync.RWMutex] which is not
|
|
||||||
// durably blocking in synctest, causing hangs.
|
// durably blocking in synctest, causing hangs.
|
||||||
// - https://github.com/golang/go/issues/77687 (mutex as durably blocking)
|
// - https://github.com/golang/go/issues/77687 (mutex as durably blocking)
|
||||||
//
|
//
|
||||||
// 3. net/http server goroutines: I/O-blocked goroutines are not durably
|
// 2. net/http server goroutines: I/O-blocked goroutines are not durably
|
||||||
// blocking, preventing bubble termination.
|
// blocking, preventing bubble termination.
|
||||||
// - https://github.com/golang/go/issues/76608 (httptest synctest support)
|
// - https://github.com/golang/go/issues/76608 (httptest synctest support)
|
||||||
func TestEphemeralNodes(t *testing.T) {
|
func TestEphemeralNodes(t *testing.T) {
|
||||||
|
|||||||
@@ -1,6 +1,10 @@
|
|||||||
package state
|
package state
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
|
"runtime"
|
||||||
|
"runtime/pprof"
|
||||||
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -10,6 +14,37 @@ import (
|
|||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func newTestAuthCache(t *testing.T, maxEntries int, ttl time.Duration) *expirable.LRU[types.AuthID, *types.AuthRequest] {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
cache := expirable.NewLRU[types.AuthID, *types.AuthRequest](
|
||||||
|
maxEntries,
|
||||||
|
func(_ types.AuthID, rn *types.AuthRequest) {
|
||||||
|
rn.FinishAuth(types.AuthVerdict{Err: ErrRegistrationExpired})
|
||||||
|
},
|
||||||
|
ttl,
|
||||||
|
)
|
||||||
|
t.Cleanup(cache.Close)
|
||||||
|
|
||||||
|
return cache
|
||||||
|
}
|
||||||
|
|
||||||
|
func requireExpiredVerdict(t *testing.T, entry *types.AuthRequest) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
select {
|
||||||
|
case verdict := <-entry.WaitForAuth():
|
||||||
|
require.False(t, verdict.Accept(), "evicted entry must not signal Accept")
|
||||||
|
require.ErrorIs(t,
|
||||||
|
verdict.Err, ErrRegistrationExpired,
|
||||||
|
"evicted entry must surface ErrRegistrationExpired, got: %v",
|
||||||
|
verdict.Err,
|
||||||
|
)
|
||||||
|
case <-time.After(5 * time.Second):
|
||||||
|
t.Fatal("eviction callback did not wake the parked AuthRequest")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// TestAuthCacheBoundedLRU verifies that the registration auth cache is
|
// TestAuthCacheBoundedLRU verifies that the registration auth cache is
|
||||||
// bounded by a maximum entry count, that exceeding the maxEntries evicts the
|
// bounded by a maximum entry count, that exceeding the maxEntries evicts the
|
||||||
// oldest entry, and that the eviction callback resolves the parked
|
// oldest entry, and that the eviction callback resolves the parked
|
||||||
@@ -17,13 +52,8 @@ import (
|
|||||||
func TestAuthCacheBoundedLRU(t *testing.T) {
|
func TestAuthCacheBoundedLRU(t *testing.T) {
|
||||||
const maxEntries = 4
|
const maxEntries = 4
|
||||||
|
|
||||||
cache := expirable.NewLRU[types.AuthID, *types.AuthRequest](
|
// Long TTL: eviction by size, not by time.
|
||||||
maxEntries,
|
cache := newTestAuthCache(t, maxEntries, time.Hour)
|
||||||
func(_ types.AuthID, rn *types.AuthRequest) {
|
|
||||||
rn.FinishAuth(types.AuthVerdict{Err: ErrRegistrationExpired})
|
|
||||||
},
|
|
||||||
time.Hour, // long TTL — we test eviction by size, not by time
|
|
||||||
)
|
|
||||||
|
|
||||||
entries := make([]*types.AuthRequest, 0, maxEntries+1)
|
entries := make([]*types.AuthRequest, 0, maxEntries+1)
|
||||||
ids := make([]types.AuthID, 0, maxEntries+1)
|
ids := make([]types.AuthID, 0, maxEntries+1)
|
||||||
@@ -43,18 +73,7 @@ func TestAuthCacheBoundedLRU(t *testing.T) {
|
|||||||
_, ok := cache.Get(ids[0])
|
_, ok := cache.Get(ids[0])
|
||||||
assert.False(t, ok, "oldest entry must be evicted when maxEntries is exceeded")
|
assert.False(t, ok, "oldest entry must be evicted when maxEntries is exceeded")
|
||||||
|
|
||||||
// The eviction callback must have woken the parked AuthRequest.
|
requireExpiredVerdict(t, entries[0])
|
||||||
select {
|
|
||||||
case verdict := <-entries[0].WaitForAuth():
|
|
||||||
require.False(t, verdict.Accept(), "evicted entry must not signal Accept")
|
|
||||||
require.ErrorIs(t,
|
|
||||||
verdict.Err, ErrRegistrationExpired,
|
|
||||||
"evicted entry must surface ErrRegistrationExpired, got: %v",
|
|
||||||
verdict.Err,
|
|
||||||
)
|
|
||||||
case <-time.After(time.Second):
|
|
||||||
t.Fatal("eviction callback did not wake the parked AuthRequest")
|
|
||||||
}
|
|
||||||
|
|
||||||
// All non-evicted entries must still be retrievable.
|
// All non-evicted entries must still be retrievable.
|
||||||
for i := 1; i <= maxEntries; i++ {
|
for i := 1; i <= maxEntries; i++ {
|
||||||
@@ -62,3 +81,70 @@ func TestAuthCacheBoundedLRU(t *testing.T) {
|
|||||||
assert.True(t, ok, "non-evicted entry %d should still be in the cache", i)
|
assert.True(t, ok, "non-evicted entry %d should still be in the cache", i)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestAuthCacheTTLExpiry verifies the TTL sweeper evicts an entry that timed
|
||||||
|
// out and that the eviction callback wakes its waiter.
|
||||||
|
func TestAuthCacheTTLExpiry(t *testing.T) {
|
||||||
|
cache := newTestAuthCache(t, 4, 50*time.Millisecond)
|
||||||
|
|
||||||
|
id := types.MustAuthID()
|
||||||
|
entry := types.NewAuthRequest()
|
||||||
|
cache.Add(id, entry)
|
||||||
|
|
||||||
|
requireExpiredVerdict(t, entry)
|
||||||
|
|
||||||
|
_, ok := cache.Get(id)
|
||||||
|
assert.False(t, ok, "expired entry must be gone from the cache")
|
||||||
|
}
|
||||||
|
|
||||||
|
func goroutineID() string {
|
||||||
|
buf := make([]byte, 64)
|
||||||
|
n := runtime.Stack(buf, false)
|
||||||
|
|
||||||
|
// "goroutine 123 [running]:"
|
||||||
|
return strings.Fields(string(buf[:n]))[1]
|
||||||
|
}
|
||||||
|
|
||||||
|
// authCacheSweepers counts expirable LRU TTL sweeper goroutines started from
|
||||||
|
// goroutine gid, so caches other tests are still tearing down do not count.
|
||||||
|
func authCacheSweepers(t *testing.T, gid string) int {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
var buf bytes.Buffer
|
||||||
|
|
||||||
|
require.NoError(t, pprof.Lookup("goroutine").WriteTo(&buf, 2))
|
||||||
|
|
||||||
|
n := 0
|
||||||
|
|
||||||
|
for g := range strings.SplitSeq(buf.String(), "\n\n") {
|
||||||
|
if strings.Contains(g, "golang-lru/v2/expirable") &&
|
||||||
|
strings.Contains(g, " in goroutine "+gid+"\n") {
|
||||||
|
n++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
return n
|
||||||
|
}
|
||||||
|
|
||||||
|
// TestStateCloseStopsAuthCacheSweeper verifies that closing a State stops
|
||||||
|
// the auth cache's TTL sweeper goroutine instead of leaking it, and that a
|
||||||
|
// request still pending at that point gets its expiry verdict rather than
|
||||||
|
// waiting forever for a sweep that never comes.
|
||||||
|
func TestStateCloseStopsAuthCacheSweeper(t *testing.T) {
|
||||||
|
gid := goroutineID()
|
||||||
|
|
||||||
|
s, err := NewState(persistTestConfig(t.TempDir() + "/headscale.db"))
|
||||||
|
require.NoError(t, err)
|
||||||
|
require.Equal(t, 1, authCacheSweepers(t, gid), "NewState must start a sweeper")
|
||||||
|
|
||||||
|
pending := types.NewAuthRequest()
|
||||||
|
s.SetAuthCacheEntry(types.MustAuthID(), pending)
|
||||||
|
|
||||||
|
require.NoError(t, s.Close())
|
||||||
|
|
||||||
|
requireExpiredVerdict(t, pending)
|
||||||
|
|
||||||
|
assert.Eventually(t, func() bool {
|
||||||
|
return authCacheSweepers(t, gid) == 0
|
||||||
|
}, 5*time.Second, 10*time.Millisecond, "sweeper must stop on State.Close")
|
||||||
|
}
|
||||||
@@ -223,14 +223,6 @@ func NewState(cfg *types.Config) (*State, error) {
|
|||||||
cacheMaxEntries = cfg.Tuning.RegisterCacheMaxEntries
|
cacheMaxEntries = cfg.Tuning.RegisterCacheMaxEntries
|
||||||
}
|
}
|
||||||
|
|
||||||
authCache := expirable.NewLRU[types.AuthID, *types.AuthRequest](
|
|
||||||
cacheMaxEntries,
|
|
||||||
func(id types.AuthID, rn *types.AuthRequest) {
|
|
||||||
rn.FinishAuth(types.AuthVerdict{Err: ErrRegistrationExpired})
|
|
||||||
},
|
|
||||||
cacheExpiration,
|
|
||||||
)
|
|
||||||
|
|
||||||
db, err := hsdb.NewHeadscaleDatabase(cfg)
|
db, err := hsdb.NewHeadscaleDatabase(cfg)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("initializing database: %w", err)
|
return nil, fmt.Errorf("initializing database: %w", err)
|
||||||
@@ -276,6 +268,15 @@ func NewState(cfg *types.Config) (*State, error) {
|
|||||||
nodeStore := NewNodeStore(nodes, policyPeersFunc(polMan), batchSize, batchTimeout)
|
nodeStore := NewNodeStore(nodes, policyPeersFunc(polMan), batchSize, batchTimeout)
|
||||||
nodeStore.Start()
|
nodeStore.Start()
|
||||||
|
|
||||||
|
// Last: NewLRU starts the TTL sweeper goroutine, which only Close stops.
|
||||||
|
authCache := expirable.NewLRU[types.AuthID, *types.AuthRequest](
|
||||||
|
cacheMaxEntries,
|
||||||
|
func(id types.AuthID, rn *types.AuthRequest) {
|
||||||
|
rn.FinishAuth(types.AuthVerdict{Err: ErrRegistrationExpired})
|
||||||
|
},
|
||||||
|
cacheExpiration,
|
||||||
|
)
|
||||||
|
|
||||||
s := &State{
|
s := &State{
|
||||||
cfg: cfg,
|
cfg: cfg,
|
||||||
|
|
||||||
@@ -327,6 +328,10 @@ func policyPeersFunc(pm policy.PolicyManager) PeersFunc {
|
|||||||
func (s *State) Close() error {
|
func (s *State) Close() error {
|
||||||
s.pings.drain()
|
s.pings.drain()
|
||||||
s.nodeStore.Stop()
|
s.nodeStore.Stop()
|
||||||
|
// Purge before Close: Close stops the sweeper, so anything still cached
|
||||||
|
// would never expire and its WaitForAuth caller would hang.
|
||||||
|
s.authCache.Purge()
|
||||||
|
s.authCache.Close()
|
||||||
|
|
||||||
err := s.db.Close()
|
err := s.db.Close()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in new issue
Block a user