Add PurgeExternalPayloads RPC for large-payload blob auto-purge

protos/backend_service.proto

@@ -57,6 +57,16 @@ service BackendService {
     // Starts a server stream for receiving work items
     rpc GetWorkItems (GetWorkItemsRequest) returns (stream WorkItem);
 
+    // Fetches up to `limit` tombstoned large-payload rows whose external blobs the worker must
+    // delete. The worker (which has storage credentials) deletes each blob, then calls
+    // AckPurgedPayloads so the backend can hard-delete those rows. Idempotent and safe under
+    // retries/duplicate callers (unacked rows stay tombstoned and are returned again).
+    rpc GetTombstonedPayloads(GetTombstonedPayloadsRequest) returns (GetTombstonedPayloadsResponse);
+
+    // Acks tombstoned payloads whose external blobs the worker has deleted; the backend hard-deletes
+    // the corresponding rows. Idempotent and safe under retries/duplicate callers.
+    rpc AckPurgedPayloads(AckPurgedPayloadsRequest) returns (AckPurgedPayloadsResponse);
+
     // Gets orchestration runtime state (history, etc.) for a given orchestration instance.
     rpc GetOrchestrationRuntimeState (GetOrchestrationRuntimeStateRequest) returns (GetOrchestrationRuntimeStateResponse);
 

protos/orchestrator_service.proto

@@ -786,6 +786,17 @@ service TaskHubSidecarService {
     rpc PurgeInstances(PurgeInstancesRequest) returns (PurgeInstancesResponse);
 
     rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem);
+
+    // Fetches up to `limit` tombstoned large-payload rows whose external blobs the worker must
+    // delete. The worker (which has storage credentials) deletes each blob, then calls
+    // AckPurgedPayloads so the backend can hard-delete those rows. Idempotent and safe under
+    // retries/duplicate callers (unacked rows stay tombstoned and are returned again).
+    rpc GetTombstonedPayloads(GetTombstonedPayloadsRequest) returns (GetTombstonedPayloadsResponse);
+
+    // Acks tombstoned payloads whose external blobs the worker has deleted; the backend hard-deletes
+    // the corresponding rows. Idempotent and safe under retries/duplicate callers.
+    rpc AckPurgedPayloads(AckPurgedPayloadsRequest) returns (AckPurgedPayloadsResponse);
+
     rpc CompleteActivityTask(ActivityResponse) returns (CompleteTaskResponse);
     rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse);
     rpc CompleteEntityTask(EntityBatchResult) returns (CompleteTaskResponse);
@@ -825,6 +836,40 @@ service TaskHubSidecarService {
     rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse);
 }
 
+// server -> client: a tombstoned payload row whose blob the worker must delete.
+message TombstonedPayload {
+    int32  partitionId = 1;
+    int64  instanceKey = 2;
+    int64  payloadId   = 3;
+    string token       = 4;   // e.g. "blob:v1:<container>:<blobName>"
+}
+
+// client -> server: ack sent after the worker deletes the blob; backend hard-deletes that row.
+message PayloadPurgeAck {
+    int32 partitionId = 1;
+    int64 instanceKey = 2;
+    int64 payloadId   = 3;
+}
+
+// client -> server: request up to `limit` tombstoned payload rows to purge.
+message GetTombstonedPayloadsRequest {
+    int32 limit = 1;
+}
+
+// server -> client: the tombstoned payload rows whose blobs the worker must delete.
+message GetTombstonedPayloadsResponse {
+    repeated TombstonedPayload payloads = 1;
+}
+
+// client -> server: acks for payloads whose blobs the worker has deleted.
+message AckPurgedPayloadsRequest {
+    repeated PayloadPurgeAck acks = 1;
+}
+
+// server -> client: response acknowledging the hard-deletes.
+message AckPurgedPayloadsResponse {
+}
+
 message GetWorkItemsRequest {
     int32 maxConcurrentOrchestrationWorkItems = 1;
     int32 maxConcurrentActivityWorkItems = 2;