Skip to content

fix(atunnel): serve stale egress policy without waiting while refreshes fail #3784

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
yjjnls wants to merge 1 commit into from
Closed

fix(atunnel): serve stale egress policy without waiting while refreshes fail #3784

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
58 changes: 58 additions & 0 deletions internal/atunnel/egresspolicy.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,9 @@ type actorEgressPolicy struct {
// logged at error level and subsequent stale-cache warnings are rate-limited.
failing bool
nextStaleWarn time.Time
// nextRetry bounds background refresh attempts while fetches keep failing
// to one per ttl.
nextRetry time.Time
}

// ActorEgressPolicy returns the source of actor's egress policy for its DNS
Expand Down Expand Up @@ -96,6 +99,16 @@ func (p *actorEgressPolicy) EgressPolicy(ctx context.Context) (*ateletpb.EgressP
if policy, ok := p.cached(); ok {
return policy, nil
}
// While refreshes fail, answer with the stale entry without waiting and
// let a background fetch retry at most once per TTL. Waiting here would
// stall the relay: it reads the actor's queries one at a time, so the
// queries queued behind one that waits egressPolicyFetchTimeout for a
// fetch that keeps failing would each wait too, and each would start
// another failing fetch.
if policy, ok := p.staleWhileFailing(); ok {
p.refreshInBackground()
return policy, nil
}
results := p.flight.DoChan("", func() (any, error) {
// A fetch may have finished since this query found the policy stale.
if policy, ok := p.cached(); ok {
Expand Down Expand Up @@ -130,6 +143,51 @@ func (p *actorEgressPolicy) cached() (*ateletpb.EgressPolicy, bool) {
return p.policy, true
}

// staleWhileFailing returns the last fetched policy while refreshes of it keep
// failing, even though the entry has expired. It is the fail-static answer the
// relay documents for an unreachable atelet, served without waiting for the
// next refresh attempt.
func (p *actorEgressPolicy) staleWhileFailing() (*ateletpb.EgressPolicy, bool) {
p.mu.Lock()
defer p.mu.Unlock()
if !p.fetched || !p.failing {
return nil, false
}
return p.policy, true
}

// refreshInBackground starts one fetch of the policy in the background, at
// most once per ttl, while queries are being answered with the stale entry.
// Its result updates the cache like any other fetch's, so recovery takes
// effect without waiting for an entry to expire.
func (p *actorEgressPolicy) refreshInBackground() {
p.mu.Lock()
now := p.now()
due := p.nextRetry.IsZero() || !now.Before(p.nextRetry)
if due {
p.nextRetry = now.Add(p.ttl)
}
p.mu.Unlock()
if !due {
return
}
go func() {
// Not canceled with any query: no query is waiting on it.
ctx := context.Background()
results := p.flight.DoChan("", func() (any, error) {
// A fetch may have finished since this retry was scheduled.
if policy, ok := p.cached(); ok {
return policy, nil
}
fetchCtx, cancel := context.WithTimeout(ctx, egressPolicyFetchTimeout)
defer cancel()
policy, err := p.fetch(fetchCtx)
return p.record(ctx, policy, err)
})
<-results
}()
}

// record updates the cache with a fetch's result and returns the policy and
// error EgressPolicy should report. When a refresh fails after a policy has
// already been fetched, record keeps and returns the stale entry (fail static),
Expand Down
177 changes: 138 additions & 39 deletions internal/atunnel/egresspolicy_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,28 +156,111 @@ func TestActorEgressPolicyRetriesAfterAFailure(t *testing.T) {
}
}

// Once a policy (or no-policy) has been cached, a failed refresh keeps serving
// the stale cache entry (fail static) until atelet responds again.
func TestActorEgressPolicyFailsStaticOnRefreshFailure(t *testing.T) {
// waitRecords waits until the source has logged want records. A refresh's
// record is the last thing its fetch updates, so this also means any
// background refresh has finished changing the cache.
func waitRecords(t *testing.T, h *recordingHandler, want int) {
t.Helper()
waitFor(t, func() bool {
h.mu.Lock()
defer h.mu.Unlock()
return len(h.records) >= want
})
}

// waitFor polls cond until it holds or 5 seconds pass.
func waitFor(t *testing.T, cond func() bool) {
t.Helper()
for i := 0; i < 5000; i++ {
if cond() {
return
}
time.Sleep(time.Millisecond)
}
t.Fatal("gave up waiting after 5s")
}

// While refreshes fail, queries are answered with the stale cache entry
// without waiting, even while the retrying fetch itself hangs; the fetch
// retries at most once per TTL in the background; and a retry that succeeds
// takes effect without a query waiting on it.
func TestActorEgressPolicyServesStaleWhileRefreshesFail(t *testing.T) {
errUnavailable := errors.New("atelet is unavailable")
policy1 := &ateletpb.EgressPolicy{Rules: []*ateletpb.EgressRule{{Http: &ateletpb.EgressDestinations{Hostnames: []string{"one.example.com"}}}}}
policy2 := &ateletpb.EgressPolicy{Rules: []*ateletpb.EgressRule{{Http: &ateletpb.EgressDestinations{Hostnames: []string{"two.example.com"}}}}}
script := &scriptedFetch{results: []fetchResult{
{policy: policy1},
{err: errUnavailable},
{err: errUnavailable},
{policy: policy2},
{policy: nil},
{err: errUnavailable},
}}
source, clock := newTestSource(script.fetch)
release := make(chan struct{})
var fetches atomic.Int32
source, clock := newTestSource(func(ctx context.Context) (*ateletpb.EgressPolicy, error) {
switch n := fetches.Add(1); {
case n == 1:
return policy1, nil
case n == 2:
return nil, errUnavailable // the first refresh fails fast
case n == 3:
<-release // later refreshes hang, like an unreachable atelet
return policy2, nil
case n == 4:
return policy2, nil // the healthy refresh after the recovery
default:
t.Error("started a fetch the test did not expect")
return policy2, nil
}
})
var logged recordingHandler
source.log = slog.New(&logged)

for i, want := range []*ateletpb.EgressPolicy{policy1, policy1, policy1, policy2, nil, nil} {
got, err := source.EgressPolicy(context.Background())
if err != nil || got != want {
t.Fatalf("step %d: EgressPolicy = %v, %v; want %v, nil", i, got, err, want)
// The initial fetch caches policy1, and the first refresh (a healthy,
// blocking one) fails and starts the failure run.
if got, err := source.EgressPolicy(context.Background()); err != nil || got != policy1 {
t.Fatalf("initial EgressPolicy = %v, %v; want %v", got, err, policy1)
}
clock.advance(egressPolicyTTL)
if got, err := source.EgressPolicy(context.Background()); err != nil || got != policy1 {
t.Fatalf("EgressPolicy after the failed refresh = %v, %v; want the stale %v", got, err, policy1)
}
waitRecords(t, &logged, 2)

// While refreshes keep failing, every query gets the stale entry without
// waiting, and only one refresh is retried per TTL.
clock.advance(egressPolicyTTL)
start := time.Now()
for i := range 3 {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
got, err := source.EgressPolicy(ctx)
cancel()
if err != nil || got != policy1 {
t.Fatalf("stale query %d = %v, %v; want %v without waiting", i, got, err, policy1)
}
clock.advance(egressPolicyTTL)
}
if elapsed := time.Since(start); elapsed > time.Second {
t.Errorf("3 queries took %s while refreshes fail; want each not to wait on a fetch", elapsed)
}
waitFor(t, func() bool { return fetches.Load() == 3 })
if n := fetches.Load(); n != 3 { // initial, failed refresh, one hung retry
t.Errorf("%d fetches after 3 stale queries within one TTL, want 3", n)
}

// The retry that succeeds replaces the stale entry without a query
// waiting for it.
close(release)
var got *ateletpb.EgressPolicy
waitFor(t, func() bool {
got, _ = source.EgressPolicy(context.Background())
return got == policy2
})
if got != policy2 {
t.Fatalf("EgressPolicy = %v after the retry succeeded; want %v", got, policy2)
}

// The recovered entry expires and refreshes block again, as in the
// healthy path.
clock.advance(egressPolicyTTL)
got, err := source.EgressPolicy(context.Background())
if err != nil || got != policy2 {
t.Fatalf("EgressPolicy after the recovery = %v, %v; want %v", got, err, policy2)
}
if n := fetches.Load(); n != 4 {
t.Errorf("%d fetches after the recovery, want 4", n)
}
}

Expand Down Expand Up @@ -289,33 +372,42 @@ func (h *recordingHandler) WithGroup(string) slog.Handler { return h }

// The source logs when the actor gains or loses its policy, an error on the
// first failure in a run, and a warning every 5 minutes while a stale cached
// policy is used.
// policy is used. While refreshes fail, the stale entry is served to queries
// without a log; the failures of the background retries retrying them once
// per TTL are what get logged.
func TestActorEgressPolicyLogsChanges(t *testing.T) {
errUnavailable := errors.New("atelet is unavailable")
policy := &ateletpb.EgressPolicy{}
steps := []struct {
advance time.Duration
result fetchResult
advance time.Duration
result fetchResult
wantRecords int
}{
// Initial fetch fails twice before any entry is cached.
{0, fetchResult{err: errUnavailable}},
{0, fetchResult{err: errUnavailable}},
{0, fetchResult{err: errUnavailable}, 1},
{0, fetchResult{err: errUnavailable}, 2},
// First successful fetch caches "no policy".
{0, fetchResult{}},
{egressPolicyTTL, fetchResult{}},
// Refresh fails: first failure logs ERROR, 1ns before 5m logs DEBUG,
// at 5m logs WARN, within the next 5m logs DEBUG, at 10m logs WARN.
{egressPolicyTTL, fetchResult{err: errUnavailable}},
{egressPolicyStaleWarnInterval - time.Nanosecond, fetchResult{err: errUnavailable}},
{time.Nanosecond, fetchResult{err: errUnavailable}},
{egressPolicyTTL, fetchResult{err: errUnavailable}},
{egressPolicyStaleWarnInterval - egressPolicyTTL, fetchResult{err: errUnavailable}},
// Recovery logs INFO and resets failure state.
{egressPolicyTTL, fetchResult{policy: policy}},
{egressPolicyTTL, fetchResult{policy: policy}},
{0, fetchResult{}, 3},
{egressPolicyTTL, fetchResult{}, 3},
// Refresh fails: the first failure of the run logs ERROR.
{egressPolicyTTL, fetchResult{err: errUnavailable}, 4},
// The background retries fail once per TTL: DEBUG, DEBUG, WARN at the
// 5m boundary, DEBUG.
{egressPolicyTTL, fetchResult{err: errUnavailable}, 5},
{egressPolicyTTL, fetchResult{err: errUnavailable}, 6},
{egressPolicyStaleWarnInterval - 2*egressPolicyTTL, fetchResult{err: errUnavailable}, 7},
{egressPolicyTTL, fetchResult{err: errUnavailable}, 8},
// A retry that recovers logs INFO and resets failure state.
{egressPolicyTTL, fetchResult{policy: policy}, 9},
{egressPolicyTTL, fetchResult{policy: policy}, 9},
// A new outage logs ERROR again on its first failure.
{egressPolicyTTL, fetchResult{err: errUnavailable}},
{egressPolicyTTL, fetchResult{}},
{egressPolicyTTL, fetchResult{err: errUnavailable}, 10},
{egressPolicyTTL, fetchResult{}, 11},
// A "no policy" entry fails static like any other while its refreshes
// fail; the run's first failure logs ERROR again, and a recovering
// retry logs INFO.
{egressPolicyTTL, fetchResult{err: errUnavailable}, 12},
{egressPolicyTTL, fetchResult{policy: policy}, 13},
}
results := make([]fetchResult, len(steps))
for i, s := range steps {
Expand All @@ -326,22 +418,29 @@ func TestActorEgressPolicyLogsChanges(t *testing.T) {
var logged recordingHandler
source.log = slog.New(&logged)

for _, s := range steps {
for i, s := range steps {
clock.advance(s.advance)
_, _ = source.EgressPolicy(context.Background())
// Once an entry (including "no policy") is cached, a failed refresh
// never turns into an error for the query.
if _, err := source.EgressPolicy(context.Background()); s.wantRecords > 3 && err != nil {
t.Errorf("step %d: EgressPolicy error = %v; want the stale entry", i, err)
}
waitRecords(t, &logged, s.wantRecords)
}
want := []string{
"ERROR Could not fetch the actor's egress policy; refusing its DNS queries",
"DEBUG Could not fetch the actor's egress policy; refusing its DNS queries",
"INFO Actor has no egress policy; blocking all DNS queries",
"ERROR Could not refresh the actor's egress policy; using stale cached policy",
"DEBUG Could not refresh the actor's egress policy; using stale cached policy",
"WARN Could not refresh the actor's egress policy; using stale cached policy",
"DEBUG Could not refresh the actor's egress policy; using stale cached policy",
"WARN Could not refresh the actor's egress policy; using stale cached policy",
"DEBUG Could not refresh the actor's egress policy; using stale cached policy",
"INFO Actor has an egress policy",
"ERROR Could not refresh the actor's egress policy; using stale cached policy",
"INFO Actor has no egress policy; blocking all DNS queries",
"ERROR Could not refresh the actor's egress policy; using stale cached policy",
"INFO Actor has an egress policy",
}
logged.mu.Lock()
defer logged.mu.Unlock()
Expand Down