diff --git a/protos/backend_service.proto b/protos/backend_service.proto index 29d19c7..22f03a9 100644 --- a/protos/backend_service.proto +++ b/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); diff --git a/protos/orchestrator_service.proto b/protos/orchestrator_service.proto index 3d9194a..e1faae5 100644 --- a/protos/orchestrator_service.proto +++ b/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::" +} + +// 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;