diff --git a/protos/orchestrator_service.proto b/protos/orchestrator_service.proto index 3d9194a..080b697 100644 --- a/protos/orchestrator_service.proto +++ b/protos/orchestrator_service.proto @@ -786,6 +786,19 @@ service TaskHubSidecarService { rpc PurgeInstances(PurgeInstancesRequest) returns (PurgeInstancesResponse); rpc GetWorkItems(GetWorkItemsRequest) returns (stream WorkItem); + + // Returns a bounded, deterministically ordered batch of due large-payload tombstones whose + // external blobs the worker must delete. Scoped to the caller's authenticated task hub. + // Only rows that are pending and whose next attempt time has arrived are returned; a row stays + // pending until its outcome is reported, so this is safe under retries and duplicate callers. + rpc GetLargePayloadTombstones(GetLargePayloadTombstonesRequest) returns (GetLargePayloadTombstonesResponse); + + // Reports the outcome of each attempted blob deletion. The backend owns retry scheduling and + // branches solely on `disposition`: it deletes rows reported as DELETED, reschedules RETRY on + // its own backoff, and moves QUARANTINED rows out of the active fetch while preserving their + // evidence. The worker never computes a retry delay. + rpc ReportLargePayloadPurgeResults(ReportLargePayloadPurgeResultsRequest) returns (ReportLargePayloadPurgeResultsResponse); + rpc CompleteActivityTask(ActivityResponse) returns (CompleteTaskResponse); rpc CompleteOrchestratorTask(OrchestratorResponse) returns (CompleteTaskResponse); rpc CompleteEntityTask(EntityBatchResult) returns (CompleteTaskResponse); @@ -825,6 +838,85 @@ service TaskHubSidecarService { rpc SkipGracefulOrchestrationTerminations(SkipGracefulOrchestrationTerminationsRequest) returns (SkipGracefulOrchestrationTerminationsResponse); } +// server -> client: one tombstoned large-payload row whose external blob the worker must delete. +message LargePayloadTombstone { + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; + + // A self-describing SDK v2 token: "blob:v2:{fullBlobUrl}". + // Legacy v1 tokens are never tombstoned: v1 carries a container name but not the storage + // account, so a delete against the configured account cannot be verified. The backend + // hard-deletes v1 payload rows instead. + string token = 4; + + // Optimistic-concurrency guard. The worker echoes this value back unmodified in the result so + // the backend can reject duplicate or stale reports without taking a per-row lease. + int64 revision = 5; +} + +// The outcome of a single blob deletion attempt. The split is by whether a failure can self-heal. +enum LargePayloadPurgeDisposition { + // Required: proto3 reserves 0 as the first value, and scalars have no field presence, so an + // unset field arrives as 0. Keeping 0 meaningless is load-bearing here: if 0 meant DELETED, a + // client that failed to set this field would make the backend delete tombstones and orphan the + // blobs permanently. The backend must reject a result carrying this value. + LARGE_PAYLOAD_PURGE_DISPOSITION_UNSPECIFIED = 0; + + // Terminal success: the tombstone is resolved and the backend deletes it. Covers the blob being + // deleted, the blob already being absent, and the blob being deliberately left in place because + // the payload store does not own it. All three are terminal because none of them can be + // improved by trying again. + LARGE_PAYLOAD_PURGE_DISPOSITION_DELETED = 1; + + // The failure may self-heal, so the row stays pending and the backend sets the next attempt. + LARGE_PAYLOAD_PURGE_DISPOSITION_RETRY = 2; + + // A deterministic failure or protocol violation that retrying can never fix. The row leaves the + // polling set but is never deleted or expired: it keeps the token, which after the payload row is + // gone is the only durable record of the blob, so discarding it would orphan the blob silently. + // Resolving a quarantined row is a deliberate operator action. Why it failed is not recorded here + // and is not meant to be; that detail lives in the worker's telemetry at full fidelity. + LARGE_PAYLOAD_PURGE_DISPOSITION_QUARANTINED = 3; +} + +// client -> server: the outcome of exactly one tombstoned row. +message LargePayloadPurgeResult { + // Row identity, echoed from the corresponding LargePayloadTombstone. + int32 partitionId = 1; + int64 instanceKey = 2; + int64 payloadId = 3; + + // Echoed unmodified from the fetched tombstone; used as a compare-and-swap guard. + int64 revision = 4; + + // The only field the backend acts on. Deliberately the only outcome field on this message: + // anything finer would be write-only. Failure detail stays in the worker's own telemetry, which + // holds the full exception rather than a lossy classification, and a row is correlated to it by + // (partitionId, instanceKey, payloadId). + LargePayloadPurgeDisposition disposition = 5; +} + +// client -> server: request up to `limit` due tombstones for the caller's task hub. +message GetLargePayloadTombstonesRequest { + // The maximum number of rows to return. The service clamps this to its own maximum. + int32 limit = 1; +} + +// server -> client: the due tombstones whose blobs the worker must delete. +message GetLargePayloadTombstonesResponse { + repeated LargePayloadTombstone tombstones = 1; +} + +// client -> server: a bounded batch of purge outcomes. +message ReportLargePayloadPurgeResultsRequest { + repeated LargePayloadPurgeResult results = 1; +} + +// server -> client: acknowledgement that the reported outcomes were recorded. +message ReportLargePayloadPurgeResultsResponse { +} + message GetWorkItemsRequest { int32 maxConcurrentOrchestrationWorkItems = 1; int32 maxConcurrentActivityWorkItems = 2; @@ -832,6 +924,16 @@ message GetWorkItemsRequest { repeated WorkerCapability capabilities = 10; WorkItemFilters workItemFilters = 11; + + // Task-hub scoped opt-in for large-payload blob auto-purge. + // + // This is a setting, not a WorkerCapability: WORKER_CAPABILITY_LARGE_PAYLOADS means a worker + // *can resolve* externalized payloads, not that the customer opted into *deleting* them. + // A capability list is presence-only and cannot express an explicit false, so a three-state + // wrapper is used instead: + // absent -> the worker has no opinion (older SDK); the persisted setting is unchanged. + // true/false -> an explicit customer choice; persisted only when the value actually differs. + google.protobuf.BoolValue large_payload_auto_purge_enabled = 12; } enum WorkerCapability {