atunnel: answer DNS from the stale egress policy while refreshes fail

internal/atunnel/dns/readme.md

@@ -69,7 +69,7 @@ UDP and TCP call up to the DNS protocol layer (`golang.org/x/net/dns/dnsmessage`
 
 `policy.go` and `evaluator.go`: DNS bypasses the egress gateway, so the relay enforces the Actor's egress policy itself.
 
-- `PolicySource` supplies the Actor's `EgressPolicy`. The ateworker passes atunnel's per-Actor source, which caches the policy for a few seconds so a policy created, updated, or deleted while the Actor runs takes effect without a restart, and keeps serving the stale cached entry if `atelet` is temporarily unreachable.
+- `PolicySource` supplies the Actor's `EgressPolicy`. The ateworker passes atunnel's per-Actor source, which caches the policy for a few seconds so a policy created, updated, or deleted while the Actor runs takes effect without a restart, and keeps serving the stale cached entry if `atelet` is temporarily unreachable. After a failed refresh it retries once per TTL, and queries don't wait for those retries.
 - `evaluator` (`evaluator.go`) compiles the hostname and wildcard patterns from an `EgressPolicy`'s `http`, `https`, and `tls_passthrough` rules and evaluates normalized `QNAME`s via `allowsHostname(qname)`.
 - `egressPolicyCheck` adapts the source to the `dnsHandler`'s `queryPolicy` hook, caching the compiled `*evaluator` for the current `*ateletpb.EgressPolicy` pointer and consulting it for every query. Fetching uses the `Server`'s context, so `Stop()` cancels a fetch in progress.
 - `allowQuery` evaluates a single `dnsmessage.Question` against a policy using `evaluator`.

internal/atunnel/egresspolicy.go

@@ -35,7 +35,7 @@ import (
 
 const (
 	// egressPolicyTTL is how long an actor's egress policy is reused before it
-	// is fetched again.
+	// is fetched again, and how long a failed refresh waits before the next.
 	//
 	// TODO: tune once the cost of fetching is known at scale.
 	egressPolicyTTL = 5 * time.Second
@@ -93,12 +93,13 @@ func (t *Tunnel) ActorEgressPolicy(actor resources.ActorAttribution) dns.PolicyS
 // Callers must not modify the policy as it may be shared in response to concurrent calls
 // to EgressPolicy.
 func (p *actorEgressPolicy) EgressPolicy(ctx context.Context) (*ateletpb.EgressPolicy, error) {
-	if policy, ok := p.cached(); ok {
+	policy, fresh, stale := p.cached()
+	if fresh {
 		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 {
+		// A fetch may have finished since this query found the policy expired.
+		if policy, fresh, _ := p.cached(); fresh {
 			return policy, nil
 		}
 		// Not canceled with the query that started the fetch: the others
@@ -108,6 +109,12 @@ func (p *actorEgressPolicy) EgressPolicy(ctx context.Context) (*ateletpb.EgressP
 		policy, err := p.fetch(fetchCtx)
 		return p.record(ctx, policy, err)
 	})
+	if stale {
+		// The last refresh failed, so this one may hang until its timeout too.
+		// Answer from the stale entry now and let the refresh finish in the
+		// background, rather than hold up every query of the actor.
+		return policy, nil
+	}
 	select {
 	case res := <-results:
 		if res.Err != nil {
@@ -120,21 +127,26 @@ func (p *actorEgressPolicy) EgressPolicy(ctx context.Context) (*ateletpb.EgressP
 	}
 }
 
-// cached returns the policy last fetched, unless it has expired.
-func (p *actorEgressPolicy) cached() (*ateletpb.EgressPolicy, bool) {
+// cached returns the policy last fetched. fresh reports that it has not
+// expired; stale reports that it has, and that the last refresh failed.
+func (p *actorEgressPolicy) cached() (policy *ateletpb.EgressPolicy, fresh, stale bool) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
-	if !p.fetched || !p.now().Before(p.expires) {
-		return nil, false
+	if !p.fetched {
+		return nil, false, false
+	}
+	if p.now().Before(p.expires) {
+		return p.policy, true, false
 	}
-	return p.policy, true
+	return p.policy, false, p.failing
 }
 
 // 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),
-// logging an error on the first failure and a warning every
-// egressPolicyStaleWarnInterval while stale entries are used.
+// already been fetched, record keeps and returns the stale entry (fail static)
+// for another TTL before the next refresh, logging an error on the first
+// failure and a warning every egressPolicyStaleWarnInterval while stale
+// entries are used.
 func (p *actorEgressPolicy) record(ctx context.Context, policy *ateletpb.EgressPolicy, err error) (*ateletpb.EgressPolicy, error) {
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -162,6 +174,8 @@ func (p *actorEgressPolicy) record(ctx context.Context, policy *ateletpb.EgressP
 			p.nextStaleWarn = now.Add(egressPolicyStaleWarnInterval)
 		}
 		p.log.LogAttrs(ctx, level, "Could not refresh the actor's egress policy; using stale cached policy", slog.Any("err", err))
+		// Try again after a TTL rather than on the next query.
+		p.expires = now.Add(p.ttl)
 		return p.policy, nil
 	}
 

internal/atunnel/egresspolicy_test.go

@@ -98,6 +98,11 @@ func newTestSource(fetch func(context.Context) (*ateletpb.EgressPolicy, error))
 	}, clock
 }
 
+// waitForRefresh returns once the fetch in flight, if any, has been recorded.
+func waitForRefresh(p *actorEgressPolicy) {
+	_, _, _ = p.flight.Do("", func() (any, error) { return nil, nil })
+}
+
 func TestActorEgressPolicyReusesAFetchForItsTTL(t *testing.T) {
 	policy := &ateletpb.EgressPolicy{Rules: []*ateletpb.EgressRule{{Http: &ateletpb.EgressDestinations{Hostnames: []string{"example.com"}}}}}
 	script := &scriptedFetch{results: []fetchResult{{policy: policy}, {policy: policy}}}
@@ -157,7 +162,9 @@ 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.
+// the stale cache entry (fail static) until atelet responds again. While
+// refreshes fail, queries don't wait for the next one, so the query that
+// starts the refresh atelet answers still gets the stale entry.
 func TestActorEgressPolicyFailsStaticOnRefreshFailure(t *testing.T) {
 	errUnavailable := errors.New("atelet is unavailable")
 	policy1 := &ateletpb.EgressPolicy{Rules: []*ateletpb.EgressRule{{Http: &ateletpb.EgressDestinations{Hostnames: []string{"one.example.com"}}}}}
@@ -172,15 +179,106 @@ func TestActorEgressPolicyFailsStaticOnRefreshFailure(t *testing.T) {
 	}}
 	source, clock := newTestSource(script.fetch)
 
-	for i, want := range []*ateletpb.EgressPolicy{policy1, policy1, policy1, policy2, nil, nil} {
+	for i, step := range []struct {
+		advance time.Duration
+		want    *ateletpb.EgressPolicy
+	}{
+		{0, policy1},
+		{egressPolicyTTL, policy1},
+		{egressPolicyTTL, policy1},
+		{egressPolicyTTL, policy1},
+		{0, policy2},
+		{egressPolicyTTL, nil},
+		{egressPolicyTTL, nil},
+	} {
+		clock.advance(step.advance)
 		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)
+		if err != nil || got != step.want {
+			t.Fatalf("step %d: EgressPolicy = %v, %v; want %v, nil", i, got, err, step.want)
 		}
-		clock.advance(egressPolicyTTL)
+		waitForRefresh(source)
+	}
+	if n := script.fetches.Load(); n != int32(len(script.results)) {
+		t.Errorf("%d fetches, want %d", n, len(script.results))
 	}
 }
 
+// After a failed refresh, queries answer from the stale entry without waiting
+// on another fetch: none starts until a TTL has passed, and one that hangs
+// holds up no query.
+func TestActorEgressPolicyDoesNotWaitOnRefreshesAfterAFailure(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"}}}}}
+	hung := make(chan struct{})
+	release := make(chan struct{})
+	var releaseOnce sync.Once
+	t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) })
+	var fetches atomic.Int32
+	source, clock := newTestSource(func(context.Context) (*ateletpb.EgressPolicy, error) {
+		switch fetches.Add(1) {
+		case 1:
+			return policy1, nil
+		case 2:
+			return nil, errUnavailable
+		case 3:
+			close(hung)
+			<-release
+			return policy2, nil
+		default:
+			return nil, errors.New("unexpected fetch")
+		}
+	})
+
+	// query fails the test if EgressPolicy waits on the hung refresh.
+	query := func(name string, want *ateletpb.EgressPolicy) {
+		t.Helper()
+		done := make(chan *ateletpb.EgressPolicy, 1)
+		go func() {
+			got, err := source.EgressPolicy(context.Background())
+			if err != nil {
+				got = nil
+			}
+			done <- got
+		}()
+		select {
+		case got := <-done:
+			if got != want {
+				t.Fatalf("%s: EgressPolicy = %v, want %v", name, got, want)
+			}
+		case <-time.After(5 * time.Second):
+			t.Fatalf("%s: the query waited on a hung refresh", name)
+		}
+	}
+	wantFetches := func(name string, want int32) {
+		t.Helper()
+		if n := fetches.Load(); n != want {
+			t.Fatalf("%s: %d fetches, want %d", name, n, want)
+		}
+	}
+
+	query("first fetch", policy1)
+	clock.advance(egressPolicyTTL)
+	query("failed refresh", policy1)
+	wantFetches("failed refresh", 2)
+	clock.advance(egressPolicyTTL - time.Nanosecond)
+	query("within a TTL of the failure", policy1)
+	wantFetches("within a TTL of the failure", 2)
+	clock.advance(time.Nanosecond)
+	query("refresh hangs", policy1)
+	select {
+	case <-hung:
+	case <-time.After(5 * time.Second):
+		t.Fatal("no refresh started once a TTL had passed after the failure")
+	}
+	query("refresh still hangs", policy1)
+	wantFetches("refresh still hangs", 3)
+	releaseOnce.Do(func() { close(release) })
+	waitForRefresh(source)
+	query("after atelet answers", policy2)
+	wantFetches("after atelet answers", 3)
+}
+
 // The queries an actor sends together wait on one fetch rather than each
 // asking atelet.
 func TestActorEgressPolicySharesOneFetch(t *testing.T) {
@@ -303,11 +401,12 @@ func TestActorEgressPolicyLogsChanges(t *testing.T) {
 		// 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.
+		// Refresh fails: the first failure logs ERROR, a retry 1ns before 5m
+		// logs DEBUG, the next retry (a TTL later, past 5m) logs WARN, the one
+		// after logs DEBUG, and one 5m after the WARN logs WARN again.
 		{egressPolicyTTL, fetchResult{err: errUnavailable}},
 		{egressPolicyStaleWarnInterval - time.Nanosecond, fetchResult{err: errUnavailable}},
-		{time.Nanosecond, fetchResult{err: errUnavailable}},
+		{egressPolicyTTL, fetchResult{err: errUnavailable}},
 		{egressPolicyTTL, fetchResult{err: errUnavailable}},
 		{egressPolicyStaleWarnInterval - egressPolicyTTL, fetchResult{err: errUnavailable}},
 		// Recovery logs INFO and resets failure state.
@@ -329,6 +428,7 @@ func TestActorEgressPolicyLogsChanges(t *testing.T) {
 	for _, s := range steps {
 		clock.advance(s.advance)
 		_, _ = source.EgressPolicy(context.Background())
+		waitForRefresh(source)
 	}
 	want := []string{
 		"ERROR Could not fetch the actor's egress policy; refusing its DNS queries",