From ccee539580c2aa3559d1b4a57513bad90c198991 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 8 Jul 2026 14:45:34 -0700 Subject: [PATCH 1/4] Add PurgeExternalPayloads RPC for large-payload blob auto-purge Adds the authoritative bidirectional streaming RPC and its two messages (TombstonedPayload, PayloadPurged) that the backend and worker use to purge externalized large-payload blobs. The backend soft-deletes blob-externalized payload rows and streams the tombstoned tokens to a connected worker (which has storage credentials); the worker deletes each blob and acks so the backend can hard-delete the row. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- protos/orchestrator_service.proto | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/protos/orchestrator_service.proto b/protos/orchestrator_service.proto index 3d9194a..a762a19 100644 --- a/protos/orchestrator_service.proto +++ b/protos/orchestrator_service.proto @@ -786,6 +786,12 @@ service TaskHubSidecarService { rpc PurgeInstances(PurgeInstancesRequest) returns (PurgeInstancesResponse); rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem); + + // Streams tombstoned large-payload blob tokens to a connected worker for deletion; the worker + // acks each deleted blob so the backend can hard-delete the corresponding row. Idempotent and + // safe with multiple worker replicas (unacked rows stay tombstoned and are re-streamed). + rpc PurgeExternalPayloads(stream PayloadPurged) returns (stream TombstonedPayload); + rpc CompleteActivityTask(ActivityResponse) returns (CompleteTaskResponse); rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse); rpc CompleteEntityTask(EntityBatchResult) returns (CompleteTaskResponse); @@ -825,6 +831,21 @@ 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::" +} + +// client -> server: ack sent after the worker deletes the blob; backend hard-deletes that row. +message PayloadPurged { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; +} + message GetWorkItemsRequest { int32 maxConcurrentOrchestrationWorkItems = 1; int32 maxConcurrentActivityWorkItems = 2; From 26213a95c70db0a44f629af0ed4af4560884d114 Mon Sep 17 00:00:00 2001 From: wangbill Date: Wed, 8 Jul 2026 15:30:17 -0700 Subject: [PATCH 2/4] Rename PayloadPurged to PayloadPurgeAck Pure rename of the client->server ack message; field numbers/types unchanged and wire-compatible. Reads correctly (the row is not yet purged when the worker sends it) and the ...Ack suffix signals the upstream direction in the bidi signature. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- protos/orchestrator_service.proto | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/protos/orchestrator_service.proto b/protos/orchestrator_service.proto index a762a19..a67bf3b 100644 --- a/protos/orchestrator_service.proto +++ b/protos/orchestrator_service.proto @@ -790,7 +790,7 @@ service TaskHubSidecarService { // Streams tombstoned large-payload blob tokens to a connected worker for deletion; the worker // acks each deleted blob so the backend can hard-delete the corresponding row. Idempotent and // safe with multiple worker replicas (unacked rows stay tombstoned and are re-streamed). - rpc PurgeExternalPayloads(stream PayloadPurged) returns (stream TombstonedPayload); + rpc PurgeExternalPayloads(stream PayloadPurgeAck) returns (stream TombstonedPayload); rpc CompleteActivityTask(ActivityResponse) returns (CompleteTaskResponse); rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse); @@ -840,7 +840,7 @@ message TombstonedPayload { } // client -> server: ack sent after the worker deletes the blob; backend hard-deletes that row. -message PayloadPurged { +message PayloadPurgeAck { int32 partitionId = 1; int64 instanceKey = 2; int64 payloadId = 3; From c28312c2ceef4ed297836f7d7abe332a60963387 Mon Sep 17 00:00:00 2001 From: wangbill Date: Thu, 9 Jul 2026 10:27:01 -0700 Subject: [PATCH 3/4] Add PurgeExternalPayloads RPC to BackendService The DTFx AzureManaged SDK dials only BackendService, so it needs to drain externalized large-payload blobs via that service. Mirrors the RPC already on TaskHubSidecarService; reuses TombstonedPayload and PayloadPurgeAck, which are defined in orchestrator_service.proto (already imported here). Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- protos/backend_service.proto | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/protos/backend_service.proto b/protos/backend_service.proto index 29d19c7..23b49a0 100644 --- a/protos/backend_service.proto +++ b/protos/backend_service.proto @@ -57,6 +57,11 @@ service BackendService { // Starts a server stream for receiving work items rpc GetWorkItems (GetWorkItemsRequest) returns (stream WorkItem); + // Streams tombstoned large-payload blob tokens to a connected worker for deletion; the worker + // acks each deleted blob so the backend can hard-delete the corresponding row. Idempotent and + // safe with multiple worker replicas (unacked rows stay tombstoned and are re-streamed). + rpc PurgeExternalPayloads(stream PayloadPurgeAck) returns (stream TombstonedPayload); + // Gets orchestration runtime state (history, etc.) for a given orchestration instance. rpc GetOrchestrationRuntimeState (GetOrchestrationRuntimeStateRequest) returns (GetOrchestrationRuntimeStateResponse); From 008f92348a10542c912a1a87df512e61e1f003d4 Mon Sep 17 00:00:00 2001 From: wangbill Date: Mon, 13 Jul 2026 11:44:41 -0700 Subject: [PATCH 4/4] Replace bidi PurgeExternalPayloads with two unary RPCs Replaces the bidirectional streaming PurgeExternalPayloads on both TaskHubSidecarService and BackendService with two unary RPCs: GetTombstonedPayloads(limit) to fetch tombstoned large-payload rows and AckPurgedPayloads(acks) to confirm blob deletion so the backend hard-deletes those rows. Adds the four request/response messages to orchestrator_service.proto; backend_service.proto reuses them via its existing import. TombstonedPayload and PayloadPurgeAck are unchanged. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- protos/backend_service.proto | 13 +++++++++---- protos/orchestrator_service.proto | 32 +++++++++++++++++++++++++++---- 2 files changed, 37 insertions(+), 8 deletions(-) diff --git a/protos/backend_service.proto b/protos/backend_service.proto index 23b49a0..22f03a9 100644 --- a/protos/backend_service.proto +++ b/protos/backend_service.proto @@ -57,10 +57,15 @@ service BackendService { // Starts a server stream for receiving work items rpc GetWorkItems (GetWorkItemsRequest) returns (stream WorkItem); - // Streams tombstoned large-payload blob tokens to a connected worker for deletion; the worker - // acks each deleted blob so the backend can hard-delete the corresponding row. Idempotent and - // safe with multiple worker replicas (unacked rows stay tombstoned and are re-streamed). - rpc PurgeExternalPayloads(stream PayloadPurgeAck) returns (stream TombstonedPayload); + // 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); diff --git a/protos/orchestrator_service.proto b/protos/orchestrator_service.proto index a67bf3b..e1faae5 100644 --- a/protos/orchestrator_service.proto +++ b/protos/orchestrator_service.proto @@ -787,10 +787,15 @@ service TaskHubSidecarService { rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem); - // Streams tombstoned large-payload blob tokens to a connected worker for deletion; the worker - // acks each deleted blob so the backend can hard-delete the corresponding row. Idempotent and - // safe with multiple worker replicas (unacked rows stay tombstoned and are re-streamed). - rpc PurgeExternalPayloads(stream PayloadPurgeAck) returns (stream TombstonedPayload); + // 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); @@ -846,6 +851,25 @@ message PayloadPurgeAck { 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;