diff --git a/coderd/cryptokeys/cache.go b/coderd/cryptokeys/cache.go index 503b715ed4..2fa1784599 100644 --- a/coderd/cryptokeys/cache.go +++ b/coderd/cryptokeys/cache.go @@ -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. diff --git a/coderd/cryptokeys/cache_test.go b/coderd/cryptokeys/cache_test.go index cd7f123e5a..b040ec5923 100644 --- a/coderd/cryptokeys/cache_test.go +++ b/coderd/cryptokeys/cache_test.go @@ -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) }