diff --git a/samples/LargePayloadConsoleApp/Program.cs b/samples/LargePayloadConsoleApp/Program.cs
index 97e53112..da293b01 100644
--- a/samples/LargePayloadConsoleApp/Program.cs
+++ b/samples/LargePayloadConsoleApp/Program.cs
@@ -129,7 +129,8 @@
return string.Empty;
}
- if (value.StartsWith("blob:v1:", StringComparison.Ordinal))
+ if (value.StartsWith("blob:v1:", StringComparison.Ordinal)
+ || value.StartsWith("blob:v2:", StringComparison.Ordinal))
{
throw new InvalidOperationException("Activity received a payload token instead of raw input.");
}
diff --git a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj
index 0a524ed6..78d80fd7 100644
--- a/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj
+++ b/src/Extensions/AzureBlobPayloads/AzureBlobPayloads.csproj
@@ -26,4 +26,8 @@
+
+
+
+
\ No newline at end of file
diff --git a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs
index e01d2e74..ee78f814 100644
--- a/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs
+++ b/src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs
@@ -13,11 +13,14 @@ namespace Microsoft.DurableTask;
///
/// Azure Blob Storage implementation of .
-/// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:<container>:<blobName>".
+/// Stores payloads as blobs and returns self-describing opaque tokens in the form
+/// "blob:v2:<fullBlobUrl>", where the URL is the blob's absolute URI including the storage account.
+/// Legacy "blob:v1:<container>:<blobName>" tokens are still recognized for read back-compatibility.
///
public sealed class BlobPayloadStore : PayloadStore
{
- const string TokenPrefix = "blob:v1:";
+ const string TokenPrefixV1 = "blob:v1:";
+ const string TokenPrefixV2 = "blob:v2:";
const string ContentEncodingGzip = "gzip";
const int MaxRetryAttempts = 8;
const int BaseDelayMs = 250;
@@ -25,6 +28,7 @@ public sealed class BlobPayloadStore : PayloadStore
const int NetworkTimeoutMinutes = 2;
readonly BlobContainerClient containerClient;
readonly LargePayloadStorageOptions options;
+ readonly BlobClientOptions clientOptions;
///
/// Initializes a new instance of the class.
@@ -48,7 +52,7 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
nameof(options));
}
- BlobClientOptions clientOptions = new()
+ this.clientOptions = new BlobClientOptions
{
Retry =
{
@@ -61,8 +65,8 @@ public BlobPayloadStore(LargePayloadStorageOptions options)
};
BlobServiceClient serviceClient = hasIdentityAuth
- ? new BlobServiceClient(options.AccountUri, options.Credential, clientOptions)
- : new BlobServiceClient(options.ConnectionString, clientOptions);
+ ? new BlobServiceClient(options.AccountUri, options.Credential, this.clientOptions)
+ : new BlobServiceClient(options.ConnectionString, this.clientOptions);
this.containerClient = serviceClient.GetBlobContainerClient(options.ContainerName);
}
@@ -105,44 +109,48 @@ public override async Task UploadAsync(string payLoad, CancellationToken
await blobStream.FlushAsync(cancellationToken);
}
- return EncodeToken(this.containerClient.Name, blobName);
+ return EncodeToken(blob.Uri);
}
///
public override async Task DownloadAsync(string token, CancellationToken cancellationToken)
{
- (string container, string name) = DecodeToken(token);
- if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal))
- {
- throw new ArgumentException("Token container does not match configured container.", nameof(token));
- }
-
- BlobClient blob = this.containerClient.GetBlobClient(name);
+ (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) = DecodeToken(token);
- try
+ if (!isV2)
{
- using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken);
- Stream contentStream = result.Content;
- bool isGzip = string.Equals(
- result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
-
- if (isGzip)
+ // v1 tokens do not carry the account, so the payload is assumed to live in the configured container.
+ if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal))
{
- using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
- using StreamReader reader = new(decompressed, Encoding.UTF8);
- return await ReadToEndAsync(reader, cancellationToken);
+ throw new ArgumentException("Token container does not match configured container.", nameof(token));
}
- using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
- return await ReadToEndAsync(uncompressedReader, cancellationToken);
+ return await DownloadFromBlobAsync(this.containerClient.GetBlobClient(name), cancellationToken);
}
- catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound)
+
+ // v2 tokens are self-describing: honor the account and container encoded in the token.
+ BlobClient blob;
+ if (this.IsConfiguredContainer(containerUri!))
+ {
+ // Same account and container as the configured store: reuse it (works with any auth mode).
+ blob = this.containerClient.GetBlobClient(name);
+ }
+ else if (this.options.Credential != null)
+ {
+ // The payload lives in a different account (e.g. the store was repointed). Identity auth can still
+ // read it as long as the credential has RBAC access to that account.
+ blob = new BlobClient(blobUri, this.options.Credential, this.clientOptions);
+ }
+ else
{
throw new PayloadStorageException(
- $"The blob '{name}' was not found in container '{container}'. " +
- "The payload may have been deleted or the container was never created.",
- ex);
+ $"The externalized payload lives in a different storage account ('{containerUri}') than the " +
+ $"currently-configured payload store ('{this.containerClient.Uri}'). Cross-account payload reads " +
+ "require identity (AAD) authentication with access to both accounts; connection-string / " +
+ "account-key credentials are account-specific and cannot read another account.");
}
+
+ return await DownloadFromBlobAsync(blob, cancellationToken);
}
///
@@ -153,7 +161,60 @@ public override bool IsKnownPayloadToken(string value)
return false;
}
- return value.StartsWith(TokenPrefix, StringComparison.Ordinal);
+ return value.StartsWith(TokenPrefixV1, StringComparison.Ordinal)
+ || value.StartsWith(TokenPrefixV2, StringComparison.Ordinal);
+ }
+
+ ///
+ /// Encodes a self-describing v2 payload token from the blob's absolute URI. The token carries the full blob
+ /// URL (including the storage account) so readers can locate the payload without relying on the currently
+ /// configured store. BlobClient.Uri contains no SAS or account key and is safe to persist.
+ ///
+ /// The absolute URI of the blob holding the payload.
+ /// An opaque payload token in the form "blob:v2:<fullBlobUrl>".
+ internal static string EncodeToken(Uri blobUri) => $"{TokenPrefixV2}{blobUri}";
+
+ ///
+ /// Decodes a payload token. Supports self-describing v2 tokens ("blob:v2:<fullBlobUrl>") and legacy v1
+ /// tokens ("blob:v1:<container>:<blobName>"), the latter for read back-compatibility.
+ ///
+ /// The payload token to decode.
+ ///
+ /// A tuple describing the token: whether it is v2, the container and blob names, and (for v2 only) the
+ /// absolute blob URI and its container-level URI.
+ ///
+ internal static (bool IsV2, string Container, string Name, Uri? BlobUri, Uri? ContainerUri) DecodeToken(
+ string token)
+ {
+ if (token.StartsWith(TokenPrefixV2, StringComparison.Ordinal))
+ {
+ string rest = token.Substring(TokenPrefixV2.Length);
+ if (!Uri.TryCreate(rest, UriKind.Absolute, out Uri? blobUri))
+ {
+ throw new ArgumentException("Invalid external payload token format.", nameof(token));
+ }
+
+ BlobUriBuilder builder = new(blobUri);
+ string container = builder.BlobContainerName;
+ string name = builder.BlobName;
+ builder.BlobName = string.Empty;
+ Uri containerUri = builder.ToUri();
+ return (true, container, name, blobUri, containerUri);
+ }
+
+ if (token.StartsWith(TokenPrefixV1, StringComparison.Ordinal))
+ {
+ string rest = token.Substring(TokenPrefixV1.Length);
+ int sep = rest.IndexOf(':');
+ if (sep <= 0 || sep >= rest.Length - 1)
+ {
+ throw new ArgumentException("Invalid external payload token format.", nameof(token));
+ }
+
+ return (false, rest.Substring(0, sep), rest.Substring(sep + 1), null, null);
+ }
+
+ throw new ArgumentException("Invalid external payload token.", nameof(token));
}
static async Task WritePayloadAsync(byte[] payloadBuffer, Stream target, CancellationToken cancellationToken)
@@ -177,22 +238,43 @@ static async Task ReadToEndAsync(StreamReader reader, CancellationToken
#endif
}
- static string EncodeToken(string container, string name) => $"blob:v1:{container}:{name}";
-
- static (string Container, string Name) DecodeToken(string token)
+ static async Task DownloadFromBlobAsync(BlobClient blob, CancellationToken cancellationToken)
{
- if (!token.StartsWith(TokenPrefix, StringComparison.Ordinal))
+ try
{
- throw new ArgumentException("Invalid external payload token.", nameof(token));
- }
+ using BlobDownloadStreamingResult result = await blob.DownloadStreamingAsync(cancellationToken: cancellationToken);
+ Stream contentStream = result.Content;
+ bool isGzip = string.Equals(
+ result.Details.ContentEncoding, ContentEncodingGzip, StringComparison.OrdinalIgnoreCase);
- string rest = token.Substring(TokenPrefix.Length);
- int sep = rest.IndexOf(':');
- if (sep <= 0 || sep >= rest.Length - 1)
+ if (isGzip)
+ {
+ using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
+ using StreamReader reader = new(decompressed, Encoding.UTF8);
+ return await ReadToEndAsync(reader, cancellationToken);
+ }
+
+ using StreamReader uncompressedReader = new(contentStream, Encoding.UTF8);
+ return await ReadToEndAsync(uncompressedReader, cancellationToken);
+ }
+ catch (RequestFailedException ex) when (ex.Status == (int)HttpStatusCode.NotFound)
{
- throw new ArgumentException("Invalid external payload token format.", nameof(token));
+ throw new PayloadStorageException(
+ $"The blob '{blob.Name}' was not found in container '{blob.BlobContainerName}'. " +
+ "The payload may have been deleted or the container was never created.",
+ ex);
}
+ }
- return (rest.Substring(0, sep), rest.Substring(sep + 1));
+ bool IsConfiguredContainer(Uri tokenContainerUri)
+ {
+ Uri configured = this.containerClient.Uri;
+ return string.Equals(tokenContainerUri.Scheme, configured.Scheme, StringComparison.OrdinalIgnoreCase)
+ && string.Equals(tokenContainerUri.Host, configured.Host, StringComparison.OrdinalIgnoreCase)
+ && tokenContainerUri.Port == configured.Port
+ && string.Equals(
+ tokenContainerUri.AbsolutePath.TrimEnd('/'),
+ configured.AbsolutePath.TrimEnd('/'),
+ StringComparison.Ordinal);
}
}
diff --git a/test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs b/test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs
new file mode 100644
index 00000000..e5ba58d7
--- /dev/null
+++ b/test/Grpc.IntegrationTests/BlobPayloadStoreTests.cs
@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.DurableTask.Grpc.Tests;
+
+public class BlobPayloadStoreTests
+{
+ [Fact]
+ public void EncodeToken_FromBlobUri_ProducesSelfDescribingV2Token()
+ {
+ // Arrange
+ Uri blobUri = new("https://myaccount.blob.core.windows.net/mycontainer/abc123def456");
+
+ // Act
+ string token = BlobPayloadStore.EncodeToken(blobUri);
+
+ // Assert
+ Assert.StartsWith("blob:v2:", token);
+ Assert.Contains("mycontainer", token);
+ Assert.Contains("abc123def456", token);
+
+ // Round-trip: decoding the encoded token yields the original container and blob name.
+ (bool isV2, string container, string name, _, _) = BlobPayloadStore.DecodeToken(token);
+ Assert.True(isV2);
+ Assert.Equal("mycontainer", container);
+ Assert.Equal("abc123def456", name);
+ }
+
+ [Theory]
+ [InlineData("https://myaccount.blob.core.windows.net/mycontainer/abc123def456", "mycontainer", "abc123def456")]
+ [InlineData("http://127.0.0.1:10000/devstoreaccount1/mycontainer/abc123def456", "mycontainer", "abc123def456")]
+ public void DecodeToken_V2Url_ParsesContainerAndBlob(string blobUrl, string expectedContainer, string expectedBlob)
+ {
+ // Arrange
+ string token = "blob:v2:" + blobUrl;
+
+ // Act
+ (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) =
+ BlobPayloadStore.DecodeToken(token);
+
+ // Assert
+ Assert.True(isV2);
+ Assert.Equal(expectedContainer, container);
+ Assert.Equal(expectedBlob, name);
+ Assert.Equal(new Uri(blobUrl), blobUri);
+ Assert.NotNull(containerUri);
+ Assert.EndsWith(expectedContainer, containerUri!.AbsolutePath);
+ }
+
+ [Fact]
+ public void DecodeToken_LegacyV1Token_ParsesContainerAndBlob()
+ {
+ // Arrange
+ string token = "blob:v1:mycontainer:abc123def456";
+
+ // Act
+ (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) =
+ BlobPayloadStore.DecodeToken(token);
+
+ // Assert
+ Assert.False(isV2);
+ Assert.Equal("mycontainer", container);
+ Assert.Equal("abc123def456", name);
+ Assert.Null(blobUri);
+ Assert.Null(containerUri);
+ }
+
+ [Fact]
+ public void IsKnownPayloadToken_RecognizesV1AndV2_RejectsOtherValues()
+ {
+ // Arrange
+ BlobPayloadStore store = CreateStore();
+
+ // Act & Assert
+ Assert.True(store.IsKnownPayloadToken("blob:v1:mycontainer:abc123"));
+ Assert.True(store.IsKnownPayloadToken("blob:v2:https://acct.blob.core.windows.net/mycontainer/abc123"));
+ Assert.False(store.IsKnownPayloadToken("just a normal payload"));
+ Assert.False(store.IsKnownPayloadToken(string.Empty));
+ }
+
+ [Fact]
+ public async Task DownloadAsync_V2TokenForDifferentAccountWithoutIdentity_ThrowsBeforeNetworkCall()
+ {
+ // Arrange: the configured store uses a connection string (account-key auth, no TokenCredential), while
+ // the token points at a different storage account. Account keys are account-specific, so cross-account
+ // reads are impossible and must fail fast (before any network call) with a clear error.
+ BlobPayloadStore store = CreateStore();
+ string token =
+ "blob:v2:https://otheraccount.blob.core.windows.net/othercontainer/" + Guid.NewGuid().ToString("N");
+
+ // Act
+ PayloadStorageException ex = await Assert.ThrowsAsync(
+ () => store.DownloadAsync(token, CancellationToken.None));
+
+ // Assert
+ Assert.Contains("different storage account", ex.Message);
+ }
+
+ static BlobPayloadStore CreateStore() =>
+ new(new LargePayloadStorageOptions("UseDevelopmentStorage=true") { ContainerName = "test" });
+}