mirror of
https://github.com/coder/coder.git
synced 2026-10-16 14:48:56 +08:00
fix(coderd): recover cache after failed key fetch (#29427)
Context or other transient errors to release cache fetching blocks. Unit tests added to verify. Linear: https://linear.app/codercom/issue/PLAT-597
This commit is contained in:
@@ -296,15 +296,21 @@ func (c *cache) cryptoKey(ctx context.Context, sequence int32) (string, interfac
|
||||
c.mu.Unlock()
|
||||
keys, err := c.cryptoKeys(ctx)
|
||||
c.mu.Lock()
|
||||
defer func() {
|
||||
// Release on success or failure.
|
||||
c.fetching = false
|
||||
c.cond.Broadcast()
|
||||
}()
|
||||
if err != nil {
|
||||
return "", nil, xerrors.Errorf("get keys: %w", err)
|
||||
}
|
||||
|
||||
c.lastFetch = c.clock.Now()
|
||||
c.refresher.Reset(refreshInterval)
|
||||
// Close stops the timer; do not re-arm it if Close ran during the fetch.
|
||||
if !c.closed {
|
||||
c.refresher.Reset(refreshInterval)
|
||||
}
|
||||
c.keys = keys
|
||||
c.fetching = false
|
||||
c.cond.Broadcast()
|
||||
|
||||
key, ok = c.key(sequence)
|
||||
if !ok {
|
||||
@@ -366,19 +372,25 @@ func (c *cache) refresh() {
|
||||
|
||||
c.mu.Unlock()
|
||||
keys, err := c.cryptoKeys(c.ctx)
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
defer func() {
|
||||
// Runs on success and failure so a transient error neither stops
|
||||
// refreshes nor blocks callers until the next restart. Close stops
|
||||
// the timer; do not re-arm it if Close ran during the fetch.
|
||||
if !c.closed {
|
||||
c.refresher.Reset(refreshInterval)
|
||||
}
|
||||
c.fetching = false
|
||||
c.cond.Broadcast()
|
||||
}()
|
||||
if err != nil {
|
||||
c.logger.Error(c.ctx, "fetch crypto keys", slog.Error(err))
|
||||
return
|
||||
}
|
||||
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
|
||||
c.lastFetch = c.clock.Now()
|
||||
c.refresher.Reset(refreshInterval)
|
||||
c.keys = keys
|
||||
c.fetching = false
|
||||
c.cond.Broadcast()
|
||||
}
|
||||
|
||||
// cryptoKeys queries the control plane for the crypto keys.
|
||||
|
||||
@@ -10,7 +10,9 @@ import (
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
"go.uber.org/goleak"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
"cdr.dev/slog/v3/sloggers/slogtest"
|
||||
"github.com/coder/coder/v2/coderd/cryptokeys"
|
||||
"github.com/coder/coder/v2/codersdk"
|
||||
"github.com/coder/coder/v2/testutil"
|
||||
@@ -441,6 +443,35 @@ func TestCryptoKeyCache(t *testing.T) {
|
||||
require.Equal(t, 3, ff.called)
|
||||
})
|
||||
|
||||
// Close stops the refresh timer. A fetch that is in flight when Close runs
|
||||
// fails (Close cancels the cache context) and must not re-arm the timer.
|
||||
t.Run("CloseDuringRefreshDoesNotRearmTimer", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
ctx = testutil.Context(t, testutil.WaitShort)
|
||||
// The failed refresh logs at error level by design.
|
||||
logger = slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
|
||||
clock = quartz.NewMock(t)
|
||||
)
|
||||
|
||||
bf := newBlockingFetcher()
|
||||
bf.results <- fetchResult{}
|
||||
cache, err := cryptokeys.NewSigningCache(ctx, logger, bf, codersdk.CryptoKeyFeatureTailnetResume, cryptokeys.WithCacheClock(clock))
|
||||
require.NoError(t, err)
|
||||
testutil.RequireReceive(ctx, t, bf.started)
|
||||
|
||||
// Fire the refresh timer and hold its fetch open.
|
||||
_, advance := clock.AdvanceNext()
|
||||
testutil.RequireReceive(ctx, t, bf.started)
|
||||
|
||||
require.NoError(t, cache.Close())
|
||||
advance.MustWait(ctx)
|
||||
|
||||
_, ok := clock.Peek()
|
||||
require.False(t, ok, "refresh timer was re-armed after Close")
|
||||
})
|
||||
|
||||
t.Run("Closed", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -485,18 +516,231 @@ func TestCryptoKeyCache(t *testing.T) {
|
||||
_, err = cache.VerifyingKey(ctx, keyID(expected))
|
||||
require.ErrorIs(t, err, cryptokeys.ErrClosed)
|
||||
})
|
||||
|
||||
// A failed on-demand fetch must not leave the cache in a permanently
|
||||
// "fetching" state. Subsequent callers must be able to trigger a new fetch
|
||||
// and receive a result rather than blocking forever.
|
||||
t.Run("FetchErrorDoesNotWedge", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
ctx = testutil.Context(t, testutil.WaitShort)
|
||||
logger = testutil.Logger(t)
|
||||
clock = quartz.NewMock(t)
|
||||
)
|
||||
|
||||
now := clock.Now().UTC()
|
||||
ff := &fakeFetcher{
|
||||
keys: []codersdk.CryptoKey{},
|
||||
}
|
||||
|
||||
cache, err := cryptokeys.NewSigningCache(ctx, logger, ff, codersdk.CryptoKeyFeatureTailnetResume, cryptokeys.WithCacheClock(clock))
|
||||
require.NoError(t, err)
|
||||
defer cache.Close()
|
||||
require.Equal(t, 1, ff.called)
|
||||
|
||||
// Simulate a request whose fetch fails, for example a transient
|
||||
// database error.
|
||||
ff.err = xerrors.New("fetch failed")
|
||||
_, err = cache.VerifyingKey(ctx, "13")
|
||||
require.Error(t, err)
|
||||
require.Equal(t, 2, ff.called)
|
||||
|
||||
// The fetcher recovers and the key now exists.
|
||||
newKey := codersdk.CryptoKey{
|
||||
Feature: codersdk.CryptoKeyFeatureTailnetResume,
|
||||
Secret: generateKey(t, 64),
|
||||
Sequence: 13,
|
||||
StartsAt: now,
|
||||
}
|
||||
ff.err = nil
|
||||
ff.keys = []codersdk.CryptoKey{newKey}
|
||||
|
||||
done := make(chan fetchResult, 1)
|
||||
go func() {
|
||||
key, err := cache.VerifyingKey(ctx, keyID(newKey))
|
||||
done <- fetchResult{key: key, err: err}
|
||||
}()
|
||||
|
||||
// Fails with "context expired" if the cache is wedged.
|
||||
res := testutil.TryReceive(ctx, t, done)
|
||||
require.NoError(t, res.err)
|
||||
require.Equal(t, decodedSecret(t, newKey), res.key)
|
||||
require.Equal(t, 3, ff.called)
|
||||
})
|
||||
|
||||
// A failed background refresh must not leave the cache in a permanently
|
||||
// "fetching" state, and the refresh timer must be re-armed so the cache
|
||||
// recovers on its own.
|
||||
t.Run("RefreshErrorDoesNotWedge", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
ctx = testutil.Context(t, testutil.WaitShort)
|
||||
// The failed refresh logs at error level by design.
|
||||
logger = slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
|
||||
clock = quartz.NewMock(t)
|
||||
)
|
||||
|
||||
now := clock.Now().UTC()
|
||||
expected := codersdk.CryptoKey{
|
||||
Feature: codersdk.CryptoKeyFeatureTailnetResume,
|
||||
Secret: generateKey(t, 64),
|
||||
Sequence: 12,
|
||||
StartsAt: now,
|
||||
}
|
||||
ff := &fakeFetcher{
|
||||
keys: []codersdk.CryptoKey{expected},
|
||||
}
|
||||
|
||||
cache, err := cryptokeys.NewSigningCache(ctx, logger, ff, codersdk.CryptoKeyFeatureTailnetResume, cryptokeys.WithCacheClock(clock))
|
||||
require.NoError(t, err)
|
||||
defer cache.Close()
|
||||
require.Equal(t, 1, ff.called)
|
||||
|
||||
// The refresh timer fires and the fetch fails.
|
||||
ff.err = xerrors.New("fetch failed")
|
||||
dur, advance := clock.AdvanceNext()
|
||||
advance.MustWait(ctx)
|
||||
require.Equal(t, time.Minute*10, dur)
|
||||
require.Equal(t, 2, ff.called)
|
||||
|
||||
// The timer must be re-armed despite the failure.
|
||||
_, ok := clock.Peek()
|
||||
require.True(t, ok, "refresh timer was not re-armed after a failed refresh")
|
||||
|
||||
// Cached keys remain usable.
|
||||
ff.err = nil
|
||||
key, err := cache.VerifyingKey(ctx, keyID(expected))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, decodedSecret(t, expected), key)
|
||||
|
||||
// An uncached key must trigger a new fetch rather than block forever.
|
||||
newKey := codersdk.CryptoKey{
|
||||
Feature: codersdk.CryptoKeyFeatureTailnetResume,
|
||||
Secret: generateKey(t, 64),
|
||||
Sequence: 13,
|
||||
StartsAt: now,
|
||||
}
|
||||
ff.keys = []codersdk.CryptoKey{expected, newKey}
|
||||
|
||||
done := make(chan fetchResult, 1)
|
||||
go func() {
|
||||
key, err := cache.VerifyingKey(ctx, keyID(newKey))
|
||||
done <- fetchResult{key: key, err: err}
|
||||
}()
|
||||
|
||||
// Fails with "context expired" if the cache is wedged.
|
||||
res := testutil.TryReceive(ctx, t, done)
|
||||
require.NoError(t, res.err)
|
||||
require.Equal(t, decodedSecret(t, newKey), res.key)
|
||||
require.Equal(t, 3, ff.called)
|
||||
})
|
||||
|
||||
// A waiter blocked on another caller's fetch must be woken when that fetch
|
||||
// fails, retry the fetch itself, and succeed. Only the caller whose fetch
|
||||
// failed observes the error.
|
||||
t.Run("WaiterRetriesAfterFetchError", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
var (
|
||||
ctx = testutil.Context(t, testutil.WaitShort)
|
||||
logger = testutil.Logger(t)
|
||||
clock = quartz.NewMock(t)
|
||||
)
|
||||
|
||||
bf := newBlockingFetcher()
|
||||
// Initial fetch in the constructor returns no keys.
|
||||
bf.results <- fetchResult{}
|
||||
cache, err := cryptokeys.NewSigningCache(ctx, logger, bf, codersdk.CryptoKeyFeatureTailnetResume, cryptokeys.WithCacheClock(clock))
|
||||
require.NoError(t, err)
|
||||
defer cache.Close()
|
||||
testutil.RequireReceive(ctx, t, bf.started)
|
||||
|
||||
// Caller A misses and starts a fetch that we hold open.
|
||||
aDone := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := cache.VerifyingKey(ctx, "13")
|
||||
aDone <- err
|
||||
}()
|
||||
testutil.RequireReceive(ctx, t, bf.started)
|
||||
|
||||
// Caller B misses and waits on A's fetch. SigningKey reads the clock
|
||||
// under the cache mutex right before waiting, so trapping that call
|
||||
// guarantees B holds the mutex until it enters cond.Wait. A cannot
|
||||
// re-acquire the mutex to finish until B is waiting.
|
||||
trap := clock.Trap().Now()
|
||||
bDone := make(chan fetchResult, 1)
|
||||
go func() {
|
||||
_, key, err := cache.SigningKey(ctx)
|
||||
bDone <- fetchResult{key: key, err: err}
|
||||
}()
|
||||
trap.MustWait(ctx).MustRelease(ctx)
|
||||
trap.Close()
|
||||
|
||||
// A's fetch fails.
|
||||
bf.results <- fetchResult{err: xerrors.New("fetch failed")}
|
||||
require.Error(t, testutil.RequireReceive(ctx, t, aDone))
|
||||
|
||||
// B must wake, retry the fetch, and get the key.
|
||||
testutil.RequireReceive(ctx, t, bf.started)
|
||||
expected := codersdk.CryptoKey{
|
||||
Feature: codersdk.CryptoKeyFeatureTailnetResume,
|
||||
Secret: generateKey(t, 64),
|
||||
Sequence: 13,
|
||||
StartsAt: clock.Now().UTC(),
|
||||
}
|
||||
bf.results <- fetchResult{keys: []codersdk.CryptoKey{expected}}
|
||||
res := testutil.RequireReceive(ctx, t, bDone)
|
||||
require.NoError(t, res.err)
|
||||
require.Equal(t, decodedSecret(t, expected), res.key)
|
||||
})
|
||||
}
|
||||
|
||||
type fakeFetcher struct {
|
||||
keys []codersdk.CryptoKey
|
||||
err error
|
||||
called int
|
||||
}
|
||||
|
||||
func (f *fakeFetcher) Fetch(_ context.Context, _ codersdk.CryptoKeyFeature) ([]codersdk.CryptoKey, error) {
|
||||
f.called++
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.keys, nil
|
||||
}
|
||||
|
||||
type fetchResult struct {
|
||||
keys []codersdk.CryptoKey
|
||||
key interface{}
|
||||
err error
|
||||
}
|
||||
|
||||
// blockingFetcher signals on started for each Fetch call and blocks until the
|
||||
// test supplies a result, letting tests interleave concurrent callers.
|
||||
type blockingFetcher struct {
|
||||
started chan struct{}
|
||||
results chan fetchResult
|
||||
}
|
||||
|
||||
func newBlockingFetcher() *blockingFetcher {
|
||||
return &blockingFetcher{
|
||||
started: make(chan struct{}, 8),
|
||||
results: make(chan fetchResult, 1),
|
||||
}
|
||||
}
|
||||
|
||||
func (f *blockingFetcher) Fetch(ctx context.Context, _ codersdk.CryptoKeyFeature) ([]codersdk.CryptoKey, error) {
|
||||
f.started <- struct{}{}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case res := <-f.results:
|
||||
return res.keys, res.err
|
||||
}
|
||||
}
|
||||
|
||||
func keyID(key codersdk.CryptoKey) string {
|
||||
return strconv.FormatInt(int64(key.Sequence), 10)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user