Skip to content

Commit 9922410

Browse files
pepamartinecpepamartinec
authored
Merge pull request #2626 from keboola/pepa/PAT-1904_authCache
PAT-1904 Fix flaky app auth config propagation
2 parents 36fe37e + b6c7e4d commit 9922410

6 files changed

Lines changed: 121 additions & 38 deletions

File tree

‎internal/pkg/service/appsproxy/dataapps/appconfig/appconfig.go‎

Lines changed: 21 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,13 @@ import (
2121
// staleCacheFallbackDuration is the maximum duration for which the old configuration of an application is used if loading new configuration is not possible.
2222
const staleCacheFallbackDuration = time.Hour
2323

24+
// refreshTimeout bounds how long a single config refresh may run.
25+
// The refresh holds the per-app lock, so this also bounds how long concurrent
26+
// requests for the same app wait for an in-flight refresh.
27+
const refreshTimeout = 30 * time.Second
28+
2429
type Loader interface {
25-
GetConfig(ctx context.Context, appID api.AppID) (config api.AppConfig, modified bool, err error)
30+
GetConfig(ctx context.Context, appID api.AppID) (config api.AppConfig, err error)
2631
}
2732

2833
type loader struct {
@@ -60,7 +65,7 @@ func NewLoader(d dependencies) Loader {
6065

6166
// GetConfig gets the AppConfig by the ID from Sandboxes Service.
6267
// It handles local caching based on the Cache-Control and ETag headers.
63-
func (l *loader) GetConfig(ctx context.Context, appID api.AppID) (out api.AppConfig, modified bool, err error) {
68+
func (l *loader) GetConfig(ctx context.Context, appID api.AppID) (out api.AppConfig, err error) {
6469
ctx, span := l.telemetry.Tracer().Start(ctx, "keboola.go.apps-proxy.appconfig.Loader.GetConfig")
6570
defer span.End(&err)
6671

@@ -76,24 +81,31 @@ func (l *loader) GetConfig(ctx context.Context, appID api.AppID) (out api.AppCon
7681
// At first, the item.expiresAt is zero, so the condition is skipped.
7782
now := l.clock.Now()
7883
if now.Before(item.expiresAt) {
79-
return item.config, false, nil
84+
return item.config, nil
8085
}
8186

8287
// Send API request with cached eTag.
8388
// At first, the item.config.ETag() is empty string.
84-
newConfig, err := l.api.GetAppConfig(appID, item.config.ETag()).Send(ctx)
89+
//
90+
// The refresh runs on a context detached from the incoming request: a client
91+
// disconnect (common during OAuth redirects) must not abort the shared cache
92+
// refresh and leave this replica serving stale config. Trace context is
93+
// preserved via WithoutCancel; the timeout bounds the per-app lock hold time.
94+
fetchCtx, cancel := context.WithTimeoutCause(context.WithoutCancel(ctx), refreshTimeout, errors.New("app config refresh timeout"))
95+
defer cancel()
96+
newConfig, err := l.api.GetAppConfig(appID, item.config.ETag()).Send(fetchCtx)
8597
if err == nil {
8698
// Cache the loaded configuration
8799
item.config = *newConfig
88100
item.ExtendExpiration(now, item.config.MaxAge())
89-
return item.config, true, nil
101+
return item.config, nil
90102
}
91103

92104
// The config hasn't been modified, extend expiration, return cached version
93105
notModifierErr := api.NotModifiedError{}
94106
if errors.As(err, &notModifierErr) {
95107
item.ExtendExpiration(now, notModifierErr.MaxAge)
96-
return item.config, false, nil
108+
return item.config, nil
97109
}
98110

99111
// Only the not found error is expected
@@ -106,21 +118,21 @@ func (l *loader) GetConfig(ctx context.Context, appID api.AppID) (out api.AppCon
106118
// The item.expiresAt may be zero, if there is no cached version, then the condition is skipped.
107119
if now.Before(item.expiresAt.Add(staleCacheFallbackDuration)) {
108120
l.logger.Warnf(ctx, `using stale cache for app "%s": %s`, appID, err.Error())
109-
return item.config, false, nil
121+
return item.config, nil
110122
}
111123
}
112124

113125
// Handle not found error
114126
if apiErr != nil && apiErr.StatusCode() == http.StatusNotFound {
115127
err = svcErrors.NewResourceNotFoundError("application", appID.String(), "stack").Wrap(err)
116-
return api.AppConfig{}, false, err
128+
return api.AppConfig{}, err
117129
}
118130

119131
// Return the error if:
120132
// - It is not found error.
121133
// - There is no cached version.
122134
// - The staleCacheFallbackDuration has been exceeded.
123-
return api.AppConfig{}, false, svcErrors.
135+
return api.AppConfig{}, svcErrors.
124136
NewServiceUnavailableError(errors.PrefixErrorf(err,
125137
`unable to load configuration for application "%s"`, appID,
126138
)).

‎internal/pkg/service/appsproxy/dataapps/appconfig/appconfig_test.go‎

Lines changed: 45 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,6 @@ type attempt struct {
2929
delay time.Duration
3030
responses []*http.Response
3131
expectedErrorCode int
32-
expectedModified bool
3332
}
3433

3534
func TestLoader_LoadConfig(t *testing.T) {
@@ -80,7 +79,6 @@ func TestLoader_LoadConfig(t *testing.T) {
8079
newResponse(t, 200, appPayload, `"etag-value"`, "max-age=60"),
8180
},
8281
expectedErrorCode: 0, // no error expected
83-
expectedModified: true,
8482
},
8583
},
8684
},
@@ -92,11 +90,9 @@ func TestLoader_LoadConfig(t *testing.T) {
9290
newResponse(t, 200, appPayload, `"etag-value"`, "max-age=60"),
9391
},
9492
expectedErrorCode: 0, // no error expected
95-
expectedModified: true,
9693
},
9794
{
9895
expectedErrorCode: 0, // no error expected
99-
expectedModified: false,
10096
},
10197
},
10298
},
@@ -108,7 +104,6 @@ func TestLoader_LoadConfig(t *testing.T) {
108104
newResponse(t, 200, appPayload, `"etag-value"`, "max-age=60"),
109105
},
110106
expectedErrorCode: 0, // no error expected
111-
expectedModified: true,
112107
},
113108
{
114109
delay: 10 * time.Minute,
@@ -117,19 +112,16 @@ func TestLoader_LoadConfig(t *testing.T) {
117112
newResponse(t, 304, map[string]any{}, `"etag-value"`, "max-age=30"),
118113
},
119114
expectedErrorCode: 0, // no error expected
120-
expectedModified: false,
121115
},
122116
{
123117
expectedErrorCode: 0, // no error expected
124-
expectedModified: false,
125118
},
126119
{
127120
delay: 31 * time.Second,
128121
responses: []*http.Response{
129122
newResponse(t, 304, map[string]any{}, `"etag-value"`, "max-age=30"),
130123
},
131124
expectedErrorCode: 0, // no error expected
132-
expectedModified: false,
133125
},
134126
},
135127
},
@@ -141,19 +133,16 @@ func TestLoader_LoadConfig(t *testing.T) {
141133
newResponse(t, 200, appPayload, `"etag-value"`, "max-age=60"),
142134
},
143135
expectedErrorCode: 0, // no error expected
144-
expectedModified: true,
145136
},
146137
{
147138
delay: 10 * time.Minute,
148139
responses: []*http.Response{
149140
newResponse(t, 200, map[string]any{"upstreamAppUrl": "http://new-app.local"}, `"etag-new-value"`, "max-age=60"),
150141
},
151142
expectedErrorCode: 0, // no error expected
152-
expectedModified: true,
153143
},
154144
{
155145
expectedErrorCode: 0, // no error expected
156-
expectedModified: false,
157146
},
158147
},
159148
},
@@ -165,7 +154,6 @@ func TestLoader_LoadConfig(t *testing.T) {
165154
newResponse(t, 200, appPayload, `"etag-value"`, "max-age=60"),
166155
},
167156
expectedErrorCode: 0, // no error expected
168-
expectedModified: true,
169157
},
170158
{
171159
delay: 10 * time.Minute,
@@ -178,7 +166,6 @@ func TestLoader_LoadConfig(t *testing.T) {
178166
newResponse(t, 500, map[string]any{}, "", ""),
179167
},
180168
expectedErrorCode: 0, // no error expected
181-
expectedModified: false,
182169
},
183170
{
184171
delay: time.Hour,
@@ -191,7 +178,6 @@ func TestLoader_LoadConfig(t *testing.T) {
191178
newResponse(t, 500, map[string]any{}, "", ""),
192179
},
193180
expectedErrorCode: 500,
194-
expectedModified: false,
195181
},
196182
},
197183
},
@@ -203,20 +189,17 @@ func TestLoader_LoadConfig(t *testing.T) {
203189
newResponse(t, 200, appPayload, `"etag-value"`, "max-age=7200"),
204190
},
205191
expectedErrorCode: 0, // no error expected
206-
expectedModified: true,
207192
},
208193
{
209194
delay: 59 * time.Minute,
210195
expectedErrorCode: 0, // no error expected
211-
expectedModified: false,
212196
},
213197
{
214198
delay: 2 * time.Minute,
215199
responses: []*http.Response{
216200
newResponse(t, 304, map[string]any{}, `"etag-value"`, "max-age=30"),
217201
},
218202
expectedErrorCode: 0, // no error expected
219-
expectedModified: false,
220203
},
221204
},
222205
},
@@ -244,7 +227,7 @@ func TestLoader_LoadConfig(t *testing.T) {
244227
httpmock.ResponderFromMultipleResponses(attempt.responses, t.Log),
245228
)
246229

247-
cfg, modified, err := loader.GetConfig(ctx, appID)
230+
cfg, err := loader.GetConfig(ctx, appID)
248231
if attempt.expectedErrorCode != 0 {
249232
require.Error(t, err)
250233
var apiErr *api.Error
@@ -255,7 +238,6 @@ func TestLoader_LoadConfig(t *testing.T) {
255238
require.NoError(t, err)
256239
assert.NotEmpty(t, cfg)
257240
}
258-
assert.Equal(t, attempt.expectedModified, modified)
259241
assert.Equal(t, len(attempt.responses), transport.GetTotalCallCount())
260242
}
261243
})
@@ -293,7 +275,7 @@ func TestLoader_LoadConfig_Race(t *testing.T) {
293275
// Load configuration 10x in parallel
294276
for range 10 {
295277
wg.Go(func() {
296-
cfg, _, err := loader.GetConfig(ctx, appID)
278+
cfg, err := loader.GetConfig(ctx, appID)
297279
require.NoError(t, err)
298280
assert.Equal(t, "http://app.local", cfg.UpstreamAppURL)
299281
counter.Add(1)
@@ -307,6 +289,49 @@ func TestLoader_LoadConfig_Race(t *testing.T) {
307289
assert.Equal(t, int64(10), counter.Load())
308290
}
309291

292+
// TestLoader_GetConfig_RefreshSurvivesRequestCancellation verifies that the config
293+
// refresh is not aborted when the originating request's context is canceled (e.g. the
294+
// browser disconnects during an OAuth redirect). The refresh must run on a context
295+
// detached from the request so the shared cache is still updated with the new config.
296+
func TestLoader_GetConfig_RefreshSurvivesRequestCancellation(t *testing.T) {
297+
t.Parallel()
298+
299+
appID := api.AppID("test")
300+
appPayload := map[string]any{
301+
"appId": appID.String(),
302+
"appName": "my-test",
303+
"projectId": "123",
304+
"upstreamAppUrl": "http://app.local",
305+
}
306+
307+
clk := clockwork.NewFakeClock()
308+
d, mock := dependencies.NewMockedServiceScope(t, t.Context(), config.New(), commonDeps.WithClock(clk))
309+
310+
transport := mock.MockedHTTPTransport()
311+
transport.RegisterResponder(
312+
http.MethodGet,
313+
fmt.Sprintf("%s/apps/%s/proxy-config", mock.TestConfig().SandboxesAPI.URL, appID),
314+
// Fail if the request context is canceled; succeed otherwise. This detects
315+
// whether the refresh runs on the (canceled) request context or a detached one.
316+
func(req *http.Request) (*http.Response, error) {
317+
if err := req.Context().Err(); err != nil {
318+
return nil, err
319+
}
320+
return newResponse(t, http.StatusOK, appPayload, `"etag-value"`, "max-age=60"), nil
321+
},
322+
)
323+
loader := d.AppConfigLoader()
324+
325+
// Simulate a client that disconnected before the refresh completes.
326+
reqCtx, cancel := context.WithCancelCause(t.Context())
327+
cancel(context.Canceled)
328+
329+
cfg, err := loader.GetConfig(reqCtx, appID)
330+
require.NoError(t, err)
331+
assert.Equal(t, "http://app.local", cfg.UpstreamAppURL)
332+
assert.Equal(t, 1, transport.GetTotalCallCount())
333+
}
334+
310335
func newResponse(t *testing.T, code int, body map[string]any, eTag string, cacheControl string) *http.Response {
311336
t.Helper()
312337

‎internal/pkg/service/appsproxy/dataapps/appconfig/middleware.go‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,6 @@ const (
2121
type AppConfigResult struct {
2222
AppID api.AppID
2323
AppConfig api.AppConfig
24-
Modified bool
2524
Err error
2625
}
2726

@@ -39,11 +38,10 @@ func Middleware(configLoader Loader, host string) middleware.Middleware {
3938
if ok {
4039
ctx := req.Context()
4140

42-
appConfig, modified, err := configLoader.GetConfig(ctx, appID)
41+
appConfig, err := configLoader.GetConfig(ctx, appID)
4342
result := AppConfigResult{
4443
AppID: appID,
4544
AppConfig: appConfig,
46-
Modified: modified,
4745
Err: err,
4846
}
4947

‎internal/pkg/service/appsproxy/dataapps/appconfig/middleware_test.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@ import (
1818

1919
type testLoader struct{}
2020

21-
func (l *testLoader) GetConfig(ctx context.Context, appID api.AppID) (out api.AppConfig, modified bool, err error) {
21+
func (l *testLoader) GetConfig(ctx context.Context, appID api.AppID) (out api.AppConfig, err error) {
2222
switch appID {
2323
case "1":
2424
return api.AppConfig{
@@ -27,17 +27,17 @@ func (l *testLoader) GetConfig(ctx context.Context, appID api.AppID) (out api.Ap
2727
AppSlug: new("app-1"),
2828
ProjectID: "1",
2929
UpstreamAppURL: "https://internal.app-1.example.com",
30-
}, false, nil
30+
}, nil
3131
case "changed":
3232
return api.AppConfig{
3333
ID: "2",
3434
Name: "App 2",
3535
AppSlug: new("app-2"),
3636
ProjectID: "2",
3737
UpstreamAppURL: "https://internal.app-2.example.com",
38-
}, true, nil
38+
}, nil
3939
default:
40-
return api.AppConfig{}, false, errors.New("error")
40+
return api.AppConfig{}, errors.New("error")
4141
}
4242
}
4343

‎internal/pkg/service/appsproxy/proxy/apphandler/manager.go‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,6 +43,15 @@ type appHandlerWrapper struct {
4343
handler http.Handler
4444
cancel context.CancelCauseFunc
4545
handlerHash string // hash of UpstreamTarget + E2BAccessToken; handler is recreated when it changes
46+
configETag string // ETag of the app config the handler was built from; handler is recreated when it changes
47+
}
48+
49+
// needsRebuild reports whether the cached handler must be recreated.
50+
// The handler is keyed on config identity (the config ETag) and the upstream
51+
// hash, so any config change is picked up on the next request without relying
52+
// on a one-shot "modified" signal from the config loader.
53+
func (w *appHandlerWrapper) needsRebuild(configETag, currentHash string) bool {
54+
return w.handler == nil || w.configETag != configETag || w.handlerHash != currentHash
4655
}
4756

4857
type dependencies interface {
@@ -94,15 +103,17 @@ func (m *Manager) HandlerFor(ctx context.Context, result appconfig.AppConfigResu
94103
return m.newErrorHandler(ctx, api.AppConfig{ID: result.AppID}, result.Err)
95104
}
96105

97-
// Create a new handler when config changed, upstream URL changed, or E2B token changed.
106+
// Create a new handler when the config changed (ETag), upstream URL changed, or E2B token changed.
98107
// Only a hash is stored so raw secrets don't linger in the wrapper.
99108
currentHash := handlerHash(m.upstreamManager.AppInfo(ctx, result.AppID))
100-
if wrapper.handler == nil || result.Modified || wrapper.handlerHash != currentHash {
109+
configETag := result.AppConfig.ETag()
110+
if wrapper.needsRebuild(configETag, currentHash) {
101111
if wrapper.cancel != nil {
102112
wrapper.cancel(errors.New("configuration changed"))
103113
}
104114
wrapper.handler, wrapper.cancel = m.newHandler(ctx, result.AppConfig)
105115
wrapper.handlerHash = currentHash
116+
wrapper.configETag = configETag
106117
}
107118

108119
return wrapper.handler

0 commit comments

Comments
 (0)