diff --git a/agent/consul/acl.go b/agent/consul/acl.go
index e8b79340ca..461c05b82e 100644
--- a/agent/consul/acl.go
+++ b/agent/consul/acl.go
@@ -199,10 +199,11 @@ type ACLResolverConfig struct {
 	// Delegate that implements some helper functionality that is server/client specific
 	Delegate ACLResolverDelegate
 
-	// AutoDisable indicates that RPC responses should be checked and if they indicate ACLs are disabled
-	// remotely then disable them locally as well. This is particularly useful for the client agent
-	// so that it can detect when the servers have gotten ACLs enabled.
-	AutoDisable bool
+	// DisableDuration is the length of time to leave ACLs disabled when an RPC
+	// request to a server indicates that the ACL system is disabled. If set to
+	// 0 then ACLs will not be disabled locally. This value is always set to 0 on
+	// Servers.
+	DisableDuration time.Duration
 
 	// ACLConfig is the configuration necessary to pass through to the acl package when creating authorizers
 	// and when authorizing access
@@ -212,7 +213,7 @@ type ACLResolverConfig struct {
 	Tokens *token.Store
 }
 
-const aclDisabledTTL = 30 * time.Second
+const aclClientDisabledTTL = 30 * time.Second
 
 // TODO: rename the fields to remove the ACL prefix
 type ACLResolverSettings struct {
@@ -292,8 +293,9 @@ type ACLResolver struct {
 
 	down acl.Authorizer
 
-	autoDisable  bool
-	disabled     time.Time
+	disableDuration time.Duration
+	disabledUntil   time.Time
+	// disabledLock synchronizes access to disabledUntil
 	disabledLock sync.RWMutex
 
 	agentMasterAuthz acl.Authorizer
@@ -364,7 +366,7 @@ func NewACLResolver(config *ACLResolverConfig) (*ACLResolver, error) {
 		delegate:         config.Delegate,
 		aclConf:          config.ACLConfig,
 		cache:            cache,
-		autoDisable:      config.AutoDisable,
+		disableDuration:  config.DisableDuration,
 		down:             down,
 		tokens:           config.Tokens,
 		agentMasterAuthz: authz,
@@ -1192,17 +1194,15 @@ func (r *ACLResolver) resolveTokenToIdentityAndRoles(token string) (structs.ACLI
 	return lastIdentity, nil, lastErr
 }
 
-func (r *ACLResolver) disableACLsWhenUpstreamDisabled(err error) error {
-	if !r.autoDisable || err == nil || !acl.IsErrDisabled(err) {
-		return err
+func (r *ACLResolver) handleACLDisabledError(err error) {
+	if r.disableDuration == 0 || err == nil || !acl.IsErrDisabled(err) {
+		return
 	}
 
-	r.logger.Debug("ACLs disabled on upstream servers, will retry", "retry_interval", aclDisabledTTL)
+	r.logger.Debug("ACLs disabled on servers, will retry", "retry_interval", r.disableDuration)
 	r.disabledLock.Lock()
-	r.disabled = time.Now().Add(aclDisabledTTL)
+	r.disabledUntil = time.Now().Add(r.disableDuration)
 	r.disabledLock.Unlock()
-
-	return err
 }
 
 func (r *ACLResolver) resolveLocallyManagedToken(token string) (structs.ACLIdentity, acl.Authorizer, bool) {
@@ -1238,14 +1238,15 @@ func (r *ACLResolver) ResolveTokenToIdentityAndAuthorizer(token string) (structs
 
 	if r.delegate.UseLegacyACLs() {
 		identity, authorizer, err := r.resolveTokenLegacy(token)
-		return identity, authorizer, r.disableACLsWhenUpstreamDisabled(err)
+		r.handleACLDisabledError(err)
+		return identity, authorizer, err
 	}
 
 	defer metrics.MeasureSince([]string{"acl", "ResolveToken"}, time.Now())
 
 	identity, policies, err := r.resolveTokenToIdentityAndPolicies(token)
 	if err != nil {
-		r.disableACLsWhenUpstreamDisabled(err)
+		r.handleACLDisabledError(err)
 		if IsACLRemoteError(err) {
 			r.logger.Error("Error resolving token", "error", err)
 			return &missingIdentity{reason: "primary-dc-down", token: token}, r.down, nil
@@ -1302,7 +1303,8 @@ func (r *ACLResolver) ResolveTokenToIdentity(token string) (structs.ACLIdentity,
 
 	if r.delegate.UseLegacyACLs() {
 		identity, _, err := r.resolveTokenLegacy(token)
-		return identity, r.disableACLsWhenUpstreamDisabled(err)
+		r.handleACLDisabledError(err)
+		return identity, err
 	}
 
 	defer metrics.MeasureSince([]string{"acl", "ResolveTokenToIdentity"}, time.Now())
@@ -1316,11 +1318,11 @@ func (r *ACLResolver) ACLsEnabled() bool {
 		return false
 	}
 
-	if r.autoDisable {
+	if r.disableDuration != 0 {
 		// Whether ACLs are disabled according to RPCs failing with a ACLs Disabled error
 		r.disabledLock.RLock()
 		defer r.disabledLock.RUnlock()
-		return !time.Now().Before(r.disabled)
+		return time.Now().After(r.disabledUntil)
 	}
 
 	return true
diff --git a/agent/consul/acl_test.go b/agent/consul/acl_test.go
index d6861fec1a..c6c2ac7c07 100644
--- a/agent/consul/acl_test.go
+++ b/agent/consul/acl_test.go
@@ -731,8 +731,8 @@ func newTestACLResolver(t *testing.T, delegate *ACLResolverTestDelegate, cb func
 			Authorizers:    4,
 			Roles:          4,
 		},
-		AutoDisable: true,
-		Delegate:    delegate,
+		DisableDuration: aclClientDisabledTTL,
+		Delegate:        delegate,
 	}
 
 	if cb != nil {
@@ -3565,7 +3565,7 @@ func TestACLResolver_AgentMaster(t *testing.T) {
 	r := newTestACLResolver(t, d, func(cfg *ACLResolverConfig) {
 		cfg.Tokens = &tokens
 		cfg.Config.NodeName = "foo"
-		cfg.AutoDisable = false
+		cfg.DisableDuration = 0
 	})
 
 	tokens.UpdateAgentMasterToken("9a184a11-5599-459e-b71a-550e5f9a5a23", token.TokenSourceConfig)
@@ -3580,3 +3580,61 @@ func TestACLResolver_AgentMaster(t *testing.T) {
 	require.Equal(t, acl.Allow, authz.NodeRead("bar", nil))
 	require.Equal(t, acl.Deny, authz.NodeWrite("bar", nil))
 }
+
+func TestACLResolver_ACLsEnabled(t *testing.T) {
+	type testCase struct {
+		name     string
+		resolver *ACLResolver
+		enabled  bool
+	}
+
+	run := func(t *testing.T, tc testCase) {
+		require.Equal(t, tc.enabled, tc.resolver.ACLsEnabled())
+	}
+
+	var testCases = []testCase{
+		{
+			name:     "config disabled",
+			resolver: &ACLResolver{},
+		},
+		{
+			name: "config enabled, disableDuration=0 (Server)",
+			resolver: &ACLResolver{
+				config: ACLResolverSettings{ACLsEnabled: true},
+			},
+			enabled: true,
+		},
+		{
+			name: "config enabled, disabled by RPC (Client)",
+			resolver: &ACLResolver{
+				config:          ACLResolverSettings{ACLsEnabled: true},
+				disableDuration: 10 * time.Second,
+				disabledUntil:   time.Now().Add(5 * time.Second),
+			},
+		},
+		{
+			name: "config enabled, past disabledUntil (Client)",
+			resolver: &ACLResolver{
+				config:          ACLResolverSettings{ACLsEnabled: true},
+				disableDuration: 10 * time.Second,
+				disabledUntil:   time.Now().Add(-5 * time.Second),
+			},
+			enabled: true,
+		},
+		{
+			name: "config enabled, no disabledUntil (Client)",
+			resolver: &ACLResolver{
+				config:          ACLResolverSettings{ACLsEnabled: true},
+				disableDuration: 10 * time.Second,
+			},
+			enabled: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			run(t, tc)
+		})
+	}
+
+}
diff --git a/agent/consul/client.go b/agent/consul/client.go
index c7e8d94c7c..57047973be 100644
--- a/agent/consul/client.go
+++ b/agent/consul/client.go
@@ -123,13 +123,13 @@ func NewClient(config *Config, deps Deps) (*Client, error) {
 
 	c.useNewACLs = 0
 	aclConfig := ACLResolverConfig{
-		Config:      config.ACLResolverSettings,
-		Delegate:    c,
-		Logger:      c.logger,
-		AutoDisable: true,
-		CacheConfig: clientACLCacheConfig,
-		ACLConfig:   newACLConfig(c.logger),
-		Tokens:      deps.Tokens,
+		Config:          config.ACLResolverSettings,
+		Delegate:        c,
+		Logger:          c.logger,
+		DisableDuration: aclClientDisabledTTL,
+		CacheConfig:     clientACLCacheConfig,
+		ACLConfig:       newACLConfig(c.logger),
+		Tokens:          deps.Tokens,
 	}
 	var err error
 	if c.acls, err = NewACLResolver(&aclConfig); err != nil {
diff --git a/agent/consul/leader_connect_ca_test.go b/agent/consul/leader_connect_ca_test.go
index 3bdb1f3a52..c69373e514 100644
--- a/agent/consul/leader_connect_ca_test.go
+++ b/agent/consul/leader_connect_ca_test.go
@@ -345,6 +345,9 @@ func TestCAManager_UpdateConfigWhileRenewIntermediate(t *testing.T) {
 }
 
 func TestCAManager_SignLeafWithExpiredCert(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
 
 	args := []struct {
 		testName              string
diff --git a/agent/consul/server.go b/agent/consul/server.go
index 97e092a1b8..90b8a02fb2 100644
--- a/agent/consul/server.go
+++ b/agent/consul/server.go
@@ -429,7 +429,6 @@ func NewServer(config *Config, flat Deps) (*Server, error) {
 		Config:      config.ACLResolverSettings,
 		Delegate:    s,
 		CacheConfig: serverACLCacheConfig,
-		AutoDisable: false,
 		Logger:      logger,
 		ACLConfig:   s.aclConfig,
 		Tokens:      flat.Tokens,