我可以使用 ConcurrentDictionary 执行原子的 try-get-or-add 操作吗？

Question

在我的 ASP.NET Core MVC Web 应用程序（使用 OIDC）中，我有一个类可以在访问者身份验证 cookie 过期之前自动刷新存储在访问者身份验证 cookie 中的access_token。

它基于 IdentityServer4 示例中的 AutomaticTokenManagementCookieEvents。在这里可用：https://github.com/IdentityServer/IdentityServer4/blob/0155beb2cea850144b6407684a2eda22e4eea3db/samples/Clients/src/MvcHybridAutomaticRefresh/AutomaticTokenManagement/AutomaticTokenManagementCookieEvents.cs

static readonly ConcurrentDictionary<String,Object> _pendingRefreshes = new ConcurrentDictionary<String,Object>();

public override async Task ValidatePrincipal( CookieValidatePrincipalContext context )
{
    DateTime accessTokenExpiresAt = GetAccessTokenExpiry( context ); // gets the 'expires_at' value from `context.Properties.GetTokens();`
    String refreshToken = GetRefreshToken( context ); // Gets the 'refresh_token' value from `context.Properties.GetTokens();`

    Boolean isExpired = DateTime.UtcNow > accessTokenExpiresAt;
    Boolean willExpireSoon = DateTime.UtcNow > accessTokenExpiresAt.Subtract( TimeSpan.FromSeconds( 60 ) );

    if( isExpired || willExpireSoon )
    {
        Boolean canRefresh = _pendingRefreshes.TryAdd( refreshToken, null );
        if( canRefresh )
        {
            try
            {
                await RefreshAccessTokenAsync( context, refreshToken );
            }
            finally
            {
                _pendingRefreshes.TryRemove( refreshToken );
            }
        }
        else
        {
            // TODO: What should happen here?
        }
    } 

}

private async Task RefreshAccessTokenAsync( CookieValidatePrincipalContext context, String refreshToken )
{
    // using IdentityModel.Client.HttpClientTokenRequestExtensions.RequestRefreshTokenAsync
    TokenResponse response = await this.httpClient.RefreshTokenAsync( refreshToken );
    if( response.IsError )
    {
        // (Error logging code here)

        if( response.Error == "invalid_grant" )
        {
            // Usually invalid_grant errors happen if the user's refresh_token has been revoked or expired
            // refresh_token expiry is separate from access_token expiry.
            // If a refresh_token has expired or been revoked the only thing to do is force the user to login again. `RejectPrincipal()` will send the user to the OIDC OP login page - though this will cause the user to lose their data if this is a POST request.
            context.RejectPrincipal();
        }
        else
        {
            // Something else bad happened. Don't invalidate the user's credentials unless they're actually expired, though.
            throw new Exception( "Unexpected error." );
        }
    }
    else
    {
        context.Properties.UpdateTokenValue( "access_token" , response.AccessToken  );
        context.Properties.UpdateTokenValue( "refresh_token", response.RefreshToken );

        DateTime newExpiresAt = DateTime.UtcNow + TimeSpan.FromSeconds( response.ExpiresIn );
        context.Properties.UpdateTokenValue( "expires_at", newExpiresAt.ToString( "o", CultureInfo.InvariantCulture ) );

        await context.HttpContext.SignInAsync( context.Principal, context.Properties );
    }
}

此代码的问题在于，如果用户的浏览器在其access_token 已过期之后同时发出两个请求，那么如果稍后在 ASP.NET Core 管道中同时执行第二个代码，用户将收到一条错误消息请求使用现已过期的access_token。

...我怎样才能得到它，以便使用过期的access_token 的第二个并发请求将await 相同的Task（来自RefreshAccessTokenAsync）？

我的想法是这样的：

1. 将_pendingRefreshes 更改为ConcurrentDictionary<String,Task<String>>。

2. 将Boolean canRefresh = _pendingRefreshes.TryAdd( refreshToken, null ); 更改为类似的内容（使用假设的TryGetOrAdd 方法）：

   Boolean addedNewTask = _pendingRefreshes
       .TryGetOrAdd(
           key: refreshToken,
           valueFactory: rt => this.RefreshTokenAsync( context, rt ),
           value: out Task task
       );
   
   if( addedNewTask )
   {
       // wait for the new access_token to be saved before continuing.
       await task;
   }
   else
   {
       if( isExpired )
       {
           // If the current request's access_token is already expired and its refresh_token is currently being refrehsed, then wait for it to finish as well, then update the access_token but only for this request's lifetime (i.e. don't call `ReplacePrincipal` or `SignInAsync`.
           await task;
       }
   }
   

问题是 ConcurrentDictionary<TKey,TValue> 没有 TryGetOrAdd 方法，我可以用它来原子地获取现有或添加新项目。

• AddOrUpdate - 不返回任何现有项目。不指示返回的值是否为现有项目。
• GetOrAdd - 不指明返回的值是否为现有项目。
• TryAdd - 不允许您使用相同的键原子地获取任何现有值。
• TryGetValue - 如果给定键没有值，则不允许您自动添加新项目。
• TryRemove - 不允许您自动添加新项目。
• TryUpdate - 不允许您添加新项目。

这可以使用lock 解决，但这会抵消使用ConcurrentDictionary 的优势。像这样的：

Task<String> task;
Boolean addedNewTask;
lock( _pendingRefreshes )
{
    Boolean taskExists = _pendingRefreshes.TryGetValue( refreshToken, out task );
    if( taskExists )
    {
        addedNewTask = false;
    }
    else
    {
        task = RefreshAccessTokenAsync( context, refreshToken );
        if( !_pendingRefreshes.TryAdd( refreshToken, task ) )
        {
            throw new InvalidOperationException( "Could not add the Task." ); // This should never happen.
        }
        addedNewTask = true;
    }
}

if( addedNewTask || isExpired )
{
    String newAccessToken = await task;
    if( isExpired )
    {
        context.Properties.UpdateTokenValue( "access_token", newAccessToken );
    }
}

...或者对于这种情况，ConcurrentDictionary 的正确用法是什么？

Answer 1

我学习了 that ConcurrentDictionary does not guarantee that valueFactory won't be invoked exactly once per key，但可能会被多次调用 - 我无法忍受，因为我的 valueFactory 是一项代价高昂的操作，具有可能对用户体验产生不利负面影响的副作用（例如，使令牌无效不必要地存储在用户的 cookie 中）。

我选择了使用lock 和Dictionary<String,Task<TokenResponse>> 的方法。我确实想知道如何改进它，我非常感谢任何反馈：

private static readonly Dictionary<String,Task<TokenResponse>> _currentlyRunningRefreshOperations = new Dictionary<String,Task<TokenResponse>>();

/// <summary>Returns <c>true</c> if a new Task was returned. Returns <c>false</c> if an existing Task was returned.</summary>
private Boolean GetOrCreateRefreshTokenTaskAsync( CookieValidatePrincipalContext context, String refreshToken, out Task<TokenResponse> task )
{
    lock( _currentlyRunningRefreshOperations )
    {
        if( _currentlyRunningRefreshOperations.TryGetValue( refreshToken, out task ) )
        {
            return false;
        }
        else
        {
            task = this.AttemptRefreshTokensAsync( context, refreshToken );
            _currentlyRunningRefreshOperations.Add( refreshToken, task );
            return true;
        }
    }
}

private async Task<TokenResponse> AttemptRefreshTokensAsync( CookieValidatePrincipalContext context, String refreshToken )
{
    try
    {
        TokenResponse response = await this.service.RefreshTokenAsync( refreshToken );
        if( response.IsError )
        {
            this.logger.LogWarning( "Encountered " + nameof(this.service.RefreshTokenAsync) + " error: {error}. Type: {errorType}. Description: {errorDesc}. refresh_token: {refreshToken}.", response.Error, response.ErrorType, response.ErrorDescription, refreshToken );

            return response;
        }
    }
    finally
    {
        lock( _currentlyRunningRefreshOperations )
        {
            _currentlyRunningRefreshOperations.Remove( refreshToken );
        }
    }
}