laskoviymishka commented on code in PR #1561:
URL: https://github.com/apache/iceberg-go/pull/1561#discussion_r3680757947
##########
catalog/rest/vended_creds.go:
##########
@@ -154,6 +156,7 @@ func (v *vendedCredentialRefresher) loadFS(ctx
context.Context) (iceio.IO, error
v.cachedIO = newIO
v.expiresAt = v.expiresAtFromConfig(config)
+ v.issuedAt = v.now()
Review Comment:
`issuedAt` is the moment of first local use, not when the server issued the
credential. For a cred that arrives in `props` and sits there a while before
the first `loadFS`, the lifetime we compute is short, so `min(5m, lifetime/2)`
clamps the buffer well below 5m — a 2m token idle for 90s gives lifetime 30s,
buffer 15s.
That's still strictly better than no buffer, and you can't really squeeze 5m
of headroom out of a token with 30s left. But it does mean the "5-minute
prefetch" the title promises silently degrades for pre-aged creds — which is
the original bug in miniature. I'd add a comment saying so, or read a
server-side issued-at field if the REST response carries one. wdyt?
##########
catalog/rest/vended_creds.go:
##########
@@ -163,12 +166,45 @@ func (v *vendedCredentialRefresher) expiredError(at
time.Time) error {
ErrVendedCredentialsExpired, v.location,
at.Format(time.RFC3339))
}
-// expired reports whether the cached IO's credentials are past their expiry. A
+func (v *vendedCredentialRefresher) needsRenewal() bool {
+ if v.fetchCreds != nil {
+ return v.shouldRefresh()
+ }
+
+ return v.expired()
+}
+
+// expired reports whether the cached IO's credentials are past their expiry,
with no safety buffer. A
Review Comment:
This line's at 84 chars now, and wedging `, with no safety buffer`
mid-sentence leaves the `A` dangling at the end — godoc renders the run-on as
the summary, and it'll likely trip lll/gocritic. I'd split it:
```go
// expired reports whether the cached IO's credentials are past their expiry.
// Unlike needsRenewal, it applies no prefetch safety buffer.
// A zero expiresAt means "never expires" — see expiresAtFromConfig.
```
##########
catalog/rest/vended_creds.go:
##########
@@ -114,7 +116,7 @@ func (v *vendedCredentialRefresher) loadFS(ctx
context.Context) (iceio.IO, error
}
defer v.mu.Release(1)
- if v.cachedIO != nil && !v.expired() {
+ if v.cachedIO != nil && !v.needsRenewal() {
Review Comment:
The goroutine that trips `shouldRefresh()` here holds the semaphore for the
whole `fetchCreds` round-trip, so under a high-fan-out scan every other caller
blocks on a credential that's still valid for up to five more minutes. Java's
`CachedSupplier` serves the cached value at prefetch time and does the refresh
in the background — only `staleTime` (hard expiry) actually blocks callers.
I don't think we need to port the background-refresh machinery, but I'd like
us to pick deliberately: either accept the serialized blocking refresh at
prefetch (simpler, no goroutine to manage) or serve the still-valid cache and
let one goroutine refresh behind it. Either's fine, but it should be a choice
rather than a side effect.
Whichever way we go, a concurrent test would help pin it down — N goroutines
arriving inside the buffer window, asserting `fetchCreds` fires at most once.
`TestVendedCredsConcurrentAccess` only covers the initial-load path today. wdyt?
##########
catalog/rest/vended_creds_test.go:
##########
@@ -336,6 +340,220 @@ func TestVendedCredsServerExpiryUsedOnRefresh(t
*testing.T) {
assert.Equal(t, serverExpiry.UnixMilli(), r.expiresAt.UnixMilli())
}
+func TestVendedCredsRefreshTriggeredWithinExpiryBuffer(t *testing.T) {
+ t.Parallel()
+
+ cases := []struct {
+ name string
+ expiresIn time.Duration
+ wantRefreshed bool
+ }{
+ {
+ name: "within buffer refreshes",
+ expiresIn: defaultVendedCredentialsExpiryBuffer -
time.Minute,
+ wantRefreshed: true,
+ },
+ {
+ name: "at buffer boundary : reuse the cache",
+ expiresIn: defaultVendedCredentialsExpiryBuffer,
+ wantRefreshed: false,
+ },
+ {
+ name: "past buffer reuses cache",
+ expiresIn: defaultVendedCredentialsExpiryBuffer +
time.Minute,
+ wantRefreshed: false,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ var callCount atomic.Int32
+ now := time.Now()
+
+ r := newTestRefresher(func(ctx context.Context, ident
[]string) (iceberg.Properties, error) {
+ callCount.Add(1)
+
+ return iceberg.Properties{}, nil
+ })
+ r.nowFunc = func() time.Time { return now }
+
+ r.cachedIO = iceio.LocalFS{}
+ r.expiresAt = now.Add(tc.expiresIn)
+
+ _, err := r.loadFS(context.Background())
+ require.NoError(t, err)
+
+ if tc.wantRefreshed {
+ assert.Equal(t, int32(1), callCount.Load(),
"credentials within the expiry buffer must be proactively refreshed")
+ } else {
+ assert.Equal(t, int32(0), callCount.Load(),
"credentials outside the expiry buffer must reuse the cache")
+ }
+ })
+ }
+}
+
+func TestVendedCredsPlanScopedNotRefusedWithinBuffer(t *testing.T) {
+ t.Parallel()
+
+ now := time.Now()
+
+ // Plan-scoped refresher: no fetchCreds, so creds can't be renewed.
+ // A cred still valid but inside the prefetch buffer must be served,
not refused.
+ r := &vendedCredentialRefresher{
+ mu: semaphore.NewWeighted(1),
+ location: "file:///tmp/test",
+ props: iceberg.Properties{},
+ nowFunc: func() time.Time { return now },
+ }
+ r.cachedIO = iceio.LocalFS{}
+ r.expiresAt = now.Add(defaultVendedCredentialsExpiryBuffer -
time.Minute)
+
+ got, err := r.loadFS(context.Background())
+ require.NoError(t, err,
+ "plan-scoped creds still within their hard expiry must not be
refused early")
+ assert.Equal(t, r.cachedIO, got)
+
+ // Past the hard expiry, with no way to renew, the load fails.
+ r.nowFunc = func() time.Time { return now.Add(time.Hour) }
+ _, err = r.loadFS(context.Background())
+ require.ErrorIs(t, err, ErrVendedCredentialsExpired)
+}
+
+func TestVendedCredsRefreshBufferClampedToLifetime(t *testing.T) {
+ t.Parallel()
+
+ now := time.Now()
+ // Short-lived (2m) renewable token.
+ // Buffer : lifetime/2 (1m), so a just-issued token is served from
cache instead of re-fetched at once.
+ shortTTL := 2 * time.Minute
+
+ var callCount atomic.Int32
+ r := newTestRefresher(func(ctx context.Context, ident []string)
(iceberg.Properties, error) {
+ callCount.Add(1)
+
+ return iceberg.Properties{
+ keyS3TokenExpiresAtMs:
strconv.FormatInt(now.Add(shortTTL).UnixMilli(), 10),
Review Comment:
`fetchCreds` returns `now.Add(shortTTL)` off the outer fixed `now`, so the
refreshed token expires at the original wall time no matter when it's fetched.
After the refresh at T+90s the new cred has ~30s of life and a 15s buffer, so a
third `loadFS` at T+105s would immediately re-trigger — invisible here only
because there's no third call.
`TestVendedCredsIssuedAtUpdatedOnRefreshPreventsSelfRetrigger` already does
this right with `clock().Add(...)`. I'd mirror that here
(`r.nowFunc().Add(shortTTL)`) and add a third call asserting no immediate
re-fetch, so the test reflects a real server minting `fetch_time + TTL`.
##########
catalog/rest/vended_creds.go:
##########
@@ -163,12 +166,45 @@ func (v *vendedCredentialRefresher) expiredError(at
time.Time) error {
ErrVendedCredentialsExpired, v.location,
at.Format(time.RFC3339))
}
-// expired reports whether the cached IO's credentials are past their expiry. A
+func (v *vendedCredentialRefresher) needsRenewal() bool {
Review Comment:
Worth a doc line here: `needsRenewal()` returning true means two different
things depending on the branch. For renewable creds (`fetchCreds != nil`) it's
"we're in the prefetch window, go refresh"; for plan-scoped creds it's "hard
expired, return the error." It reads cleanly from `loadFS`, but a one-liner
spelling out the dual meaning would save the next reader the dispatch trace.
##########
catalog/rest/vended_creds.go:
##########
@@ -163,12 +166,45 @@ func (v *vendedCredentialRefresher) expiredError(at
time.Time) error {
ErrVendedCredentialsExpired, v.location,
at.Format(time.RFC3339))
}
-// expired reports whether the cached IO's credentials are past their expiry. A
+func (v *vendedCredentialRefresher) needsRenewal() bool {
+ if v.fetchCreds != nil {
+ return v.shouldRefresh()
+ }
+
+ return v.expired()
+}
+
+// expired reports whether the cached IO's credentials are past their expiry,
with no safety buffer. A
// zero expiresAt means "never expires" — see expiresAtFromConfig.
func (v *vendedCredentialRefresher) expired() bool {
return !v.expiresAt.IsZero() && v.now().After(v.expiresAt)
}
+func (v *vendedCredentialRefresher) shouldRefresh() bool {
+ if v.expiresAt.IsZero() {
+ return false
+ }
+
+ return v.now().After(v.expiresAt.Add(-v.refreshBuffer()))
+}
+
+func (v *vendedCredentialRefresher) refreshBuffer() time.Duration {
+ buffer := defaultVendedCredentialsExpiryBuffer
+ if v.issuedAt.IsZero() {
+ return buffer
+ }
+
+ half := v.expiresAt.Sub(v.issuedAt) / 2
+ if half < 0 {
+ half = 0
+ }
+ if half < buffer {
Review Comment:
Minor: this manual clamp can just use the builtin, which the package already
leans on (`min(sleep, maxDelay)` in `scan_planning.go`):
```go
buffer = min(buffer, half)
```
gocritic's minMax will probably flag the `if` form anyway.
##########
catalog/rest/vended_creds_test.go:
##########
@@ -336,6 +340,220 @@ func TestVendedCredsServerExpiryUsedOnRefresh(t
*testing.T) {
assert.Equal(t, serverExpiry.UnixMilli(), r.expiresAt.UnixMilli())
}
+func TestVendedCredsRefreshTriggeredWithinExpiryBuffer(t *testing.T) {
+ t.Parallel()
+
+ cases := []struct {
+ name string
+ expiresIn time.Duration
+ wantRefreshed bool
+ }{
+ {
+ name: "within buffer refreshes",
+ expiresIn: defaultVendedCredentialsExpiryBuffer -
time.Minute,
+ wantRefreshed: true,
+ },
+ {
+ name: "at buffer boundary : reuse the cache",
+ expiresIn: defaultVendedCredentialsExpiryBuffer,
+ wantRefreshed: false,
+ },
+ {
+ name: "past buffer reuses cache",
+ expiresIn: defaultVendedCredentialsExpiryBuffer +
time.Minute,
+ wantRefreshed: false,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ var callCount atomic.Int32
+ now := time.Now()
+
+ r := newTestRefresher(func(ctx context.Context, ident
[]string) (iceberg.Properties, error) {
+ callCount.Add(1)
+
+ return iceberg.Properties{}, nil
+ })
+ r.nowFunc = func() time.Time { return now }
+
+ r.cachedIO = iceio.LocalFS{}
+ r.expiresAt = now.Add(tc.expiresIn)
+
+ _, err := r.loadFS(context.Background())
+ require.NoError(t, err)
+
+ if tc.wantRefreshed {
+ assert.Equal(t, int32(1), callCount.Load(),
"credentials within the expiry buffer must be proactively refreshed")
+ } else {
+ assert.Equal(t, int32(0), callCount.Load(),
"credentials outside the expiry buffer must reuse the cache")
+ }
+ })
+ }
+}
+
+func TestVendedCredsPlanScopedNotRefusedWithinBuffer(t *testing.T) {
+ t.Parallel()
+
+ now := time.Now()
+
+ // Plan-scoped refresher: no fetchCreds, so creds can't be renewed.
+ // A cred still valid but inside the prefetch buffer must be served,
not refused.
+ r := &vendedCredentialRefresher{
+ mu: semaphore.NewWeighted(1),
+ location: "file:///tmp/test",
+ props: iceberg.Properties{},
+ nowFunc: func() time.Time { return now },
+ }
+ r.cachedIO = iceio.LocalFS{}
+ r.expiresAt = now.Add(defaultVendedCredentialsExpiryBuffer -
time.Minute)
+
+ got, err := r.loadFS(context.Background())
+ require.NoError(t, err,
+ "plan-scoped creds still within their hard expiry must not be
refused early")
+ assert.Equal(t, r.cachedIO, got)
Review Comment:
Since you mutate `nowFunc` and call `loadFS` again right after this, I'd
make this `require.Equal` — if it ever fires, the non-fatal assert lets the
second call run on state you didn't expect and muddies the failure. `LocalFS{}`
is a value type, so `Equal` is the right comparison here.
##########
catalog/rest/vended_creds_test.go:
##########
@@ -336,6 +340,220 @@ func TestVendedCredsServerExpiryUsedOnRefresh(t
*testing.T) {
assert.Equal(t, serverExpiry.UnixMilli(), r.expiresAt.UnixMilli())
}
+func TestVendedCredsRefreshTriggeredWithinExpiryBuffer(t *testing.T) {
+ t.Parallel()
+
+ cases := []struct {
+ name string
+ expiresIn time.Duration
+ wantRefreshed bool
+ }{
+ {
+ name: "within buffer refreshes",
+ expiresIn: defaultVendedCredentialsExpiryBuffer -
time.Minute,
+ wantRefreshed: true,
+ },
+ {
+ name: "at buffer boundary : reuse the cache",
+ expiresIn: defaultVendedCredentialsExpiryBuffer,
+ wantRefreshed: false,
+ },
+ {
+ name: "past buffer reuses cache",
+ expiresIn: defaultVendedCredentialsExpiryBuffer +
time.Minute,
+ wantRefreshed: false,
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ var callCount atomic.Int32
+ now := time.Now()
+
+ r := newTestRefresher(func(ctx context.Context, ident
[]string) (iceberg.Properties, error) {
+ callCount.Add(1)
+
+ return iceberg.Properties{}, nil
+ })
+ r.nowFunc = func() time.Time { return now }
+
+ r.cachedIO = iceio.LocalFS{}
+ r.expiresAt = now.Add(tc.expiresIn)
Review Comment:
This whole table leaves `r.issuedAt` at zero, so `refreshBuffer()` always
takes the `issuedAt.IsZero()` early return and the boundary lands at a flat 5m.
In production `issuedAt` is set after the first `loadFS`, so a 5m-TTL token
clamps to `min(5m, 2.5m) = 2.5m` and the reuse/refresh boundary sits at 2.5m,
not the 5m this case asserts.
The clamp path itself is covered in `TestVendedCredsRefreshBuffer`, so it's
not a hole in coverage overall — just this table testing the zero-`issuedAt`
boundary rather than the production one. A parallel sub-test seeding
`r.issuedAt = now` with a 10m TTL would exercise the real boundary.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
