Make blob payload delete path token-v2 aware to prevent orphaned-payload data loss

The auto-purge delete path in BlobPayloadStore was v1-only. Once self-describing
`blob:v2:<fullBlobUrl>` tokens ship (PR #766), a v2 token would hit the v1-only
DecodeToken, throw ArgumentException, and DeleteExternalBlobActivity would classify
it as a poison token -> Discarded -> the backend acks/clears the tombstone while the
blob is never deleted -> orphaned payload (leak).

Converge the shared token code with #766 verbatim (trivial future merge) and make the
delete path v2-aware, mirroring DownloadAsync resolution:

- Recognize both blob:v1: and blob:v2: prefixes; DecodeToken returns a 5-tuple and
  parses v2 URLs via BlobUriBuilder (DNS and Azurite path-style); v1 stays read-compatible.
- UploadAsync now emits v2 tokens (EncodeToken(blob.Uri)).
- DeleteAsync: v1 container-match check preserved; v2 same-account fast-path via
  IsConfiguredContainer; v2 cross-account with an identity credential deletes directly;
  v2 cross-account WITHOUT identity throws PayloadStorageException. DeleteIfExistsAsync
  keeps deletes idempotent.

Poison classification is unchanged and verified: PayloadStorageException derives from
InvalidOperationException (not ArgumentException), so a cross-account-without-identity
delete falls through DeleteExternalBlobActivity's generic catch -> Retry (stays
tombstoned until credentials are fixed), never Discarded.

Tests: mirror #766 v2 encode/round-trip, DecodeToken v2 (DNS + Azurite) and v1
back-compat, and IsKnownPayloadToken v1+v2; add a v2 same-account happy-path delete and
a v2 cross-account-without-identity DeleteAsync -> PayloadStorageException guard (throws
before any network call); add activity tests asserting ArgumentException -> Discarded,
PayloadStorageException -> Retry, and success -> Deleted.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: 44c27836-c49c-45fd-ae2b-3309f9c3f0f0

Author: YunchuWang

src/Extensions/AzureBlobPayloads/PayloadStore/BlobPayloadStore.cs

@@ -13,18 +13,22 @@ namespace Microsoft.DurableTask;
 
 /// <summary>
 /// Azure Blob Storage implementation of <see cref="PayloadStore"/>.
-/// Stores payloads as blobs and returns opaque tokens in the form "blob:v1:&lt;container&gt;:&lt;blobName&gt;".
+/// Stores payloads as blobs and returns self-describing opaque tokens in the form
+/// "blob:v2:&lt;fullBlobUrl&gt;", where the URL is the blob's absolute URI including the storage account.
+/// Legacy "blob:v1:&lt;container&gt;:&lt;blobName&gt;" tokens are still recognized for read back-compatibility.
 /// </summary>
 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;
     const int MaxDelayMs = 10_000;
     const int NetworkTimeoutMinutes = 2;
     readonly BlobContainerClient containerClient;
     readonly LargePayloadStorageOptions options;
+    readonly BlobClientOptions clientOptions;
 
     /// <summary>
     /// Initializes a new instance of the <see cref="BlobPayloadStore"/> 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);
     }
@@ -117,56 +121,85 @@ public override async Task<string> UploadAsync(string payLoad, CancellationToken
             await blobStream.FlushAsync(cancellationToken);
         }
 
-        return EncodeToken(this.containerClient.Name, blobName);
+        return EncodeToken(blob.Uri);
     }
 
     /// <inheritdoc/>
     public override async Task<string> 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);
     }
 
     /// <inheritdoc/>
     public override async Task DeleteAsync(string token, CancellationToken cancellationToken)
     {
-        (string container, string name) = DecodeToken(token);
-        if (!string.Equals(container, this.containerClient.Name, StringComparison.Ordinal))
+        (bool isV2, string container, string name, Uri? blobUri, Uri? containerUri) = DecodeToken(token);
+
+        BlobClient blob;
+        if (!isV2)
         {
-            throw new ArgumentException("Token container does not match configured container.", nameof(token));
-        }
+            // 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))
+            {
+                throw new ArgumentException("Token container does not match configured container.", nameof(token));
+            }
 
-        BlobClient blob = this.containerClient.GetBlobClient(name);
+            blob = this.containerClient.GetBlobClient(name);
+        }
+        else 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
+            // delete 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 externalized payload lives in a different storage account ('{containerUri}') than the " +
+                $"currently-configured payload store ('{this.containerClient.Uri}'). Cross-account payload deletes " +
+                "require identity (AAD) authentication with access to both accounts; connection-string / " +
+                "account-key credentials are account-specific and cannot delete from another account.");
+        }
 
         // Idempotent by design: DeleteIfExistsAsync returns false (rather than throwing) when the blob is
         // already gone, so re-delivered tombstones and concurrent purges from multiple worker replicas are safe.
@@ -184,7 +217,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);
+    }
+
+    /// <summary>
+    /// 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. <c>BlobClient.Uri</c> contains no SAS or account key and is safe to persist.
+    /// </summary>
+    /// <param name="blobUri">The absolute URI of the blob holding the payload.</param>
+    /// <returns>An opaque payload token in the form "blob:v2:&lt;fullBlobUrl&gt;".</returns>
+    internal static string EncodeToken(Uri blobUri) => $"{TokenPrefixV2}{blobUri}";
+
+    /// <summary>
+    /// Decodes a payload token. Supports self-describing v2 tokens ("blob:v2:&lt;fullBlobUrl&gt;") and legacy v1
+    /// tokens ("blob:v1:&lt;container&gt;:&lt;blobName&gt;"), the latter for read back-compatibility.
+    /// </summary>
+    /// <param name="token">The payload token to decode.</param>
+    /// <returns>
+    /// 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.
+    /// </returns>
+    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)
@@ -208,22 +294,43 @@ static async Task<string> 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<string> 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);
+
+            if (isGzip)
+            {
+                using GZipStream decompressed = new(contentStream, CompressionMode.Decompress);
+                using StreamReader reader = new(decompressed, Encoding.UTF8);
+                return await ReadToEndAsync(reader, cancellationToken);
+            }
 
-        string rest = token.Substring(TokenPrefix.Length);
-        int sep = rest.IndexOf(':');
-        if (sep <= 0 || sep >= rest.Length - 1)
+            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);
     }
-}
+}

test/AzureBlobPayloads.Tests/AutoPurge/DeleteExternalBlobActivityTests.cs

@@ -38,6 +38,53 @@ public async Task RunAsync_WhenDeleteThrowsRequestFailedNon400_LeavesTombstonedF
         result.Should().Be(BlobDeleteResult.Retry);
     }
 
+    [Fact]
+    public async Task RunAsync_WhenDeleteThrowsArgumentException_DiscardsPoisonToken()
+    {
+        // Arrange - a malformed / unknown-prefix token makes the store throw ArgumentException at decode time,
+        // before any network call. It can never succeed, so it is discarded (acked) as poison.
+        StubPayloadStore store = new(new ArgumentException("Invalid external payload token."));
+        DeleteExternalBlobActivity activity = new(store, new TestLogger<DeleteExternalBlobActivity>());
+
+        // Act
+        BlobDeleteResult result = await activity.RunAsync(null!, "not-a-valid-token");
+
+        // Assert - discarded so the backend clears the row instead of re-streaming poison forever.
+        result.Should().Be(BlobDeleteResult.Discarded);
+    }
+
+    [Fact]
+    public async Task RunAsync_WhenDeleteThrowsPayloadStorageException_LeavesTombstonedForRetry()
+    {
+        // Arrange - a v2 cross-account token without an identity credential makes the store throw
+        // PayloadStorageException. That is a permission/capability gap, not a poison token: dropping the blob
+        // would orphan it, so it must stay tombstoned for retry - the very data loss this path guards against.
+        StubPayloadStore store = new(new PayloadStorageException("cross-account delete requires identity"));
+        DeleteExternalBlobActivity activity = new(store, new TestLogger<DeleteExternalBlobActivity>());
+
+        // Act
+        BlobDeleteResult result = await activity.RunAsync(
+            null!, "blob:v2:https://otheraccount.blob.core.windows.net/c/abc123");
+
+        // Assert - retried (not discarded) so the tombstone survives until the credential is fixed.
+        result.Should().Be(BlobDeleteResult.Retry);
+    }
+
+    [Fact]
+    public async Task RunAsync_WhenDeleteSucceeds_AcksAsDeleted()
+    {
+        // Arrange - a v2 token that deletes cleanly (same-account, or cross-account with identity access).
+        StubPayloadStore store = new(deleteError: null);
+        DeleteExternalBlobActivity activity = new(store, new TestLogger<DeleteExternalBlobActivity>());
+
+        // Act
+        BlobDeleteResult result = await activity.RunAsync(
+            null!, "blob:v2:https://acct.blob.core.windows.net/c/abc123");
+
+        // Assert
+        result.Should().Be(BlobDeleteResult.Deleted);
+    }
+
     sealed class StubPayloadStore : PayloadStore
     {
         readonly Exception? deleteError;