1use std::{
4 collections::{BTreeMap, BTreeSet},
5 future::Future,
6 pin::Pin,
7 sync::Arc,
8};
9
10use runifold_core::CheckpointId;
11use thiserror::Error;
12
13use crate::{
14 LeaseDuration, WorkerId, WorkflowStoreError, WorkflowTaskCleanupLease, WorkflowTaskLegalHold,
15 WorkflowTaskLegalHoldReason, WorkflowTaskTombstone, WorkflowTaskTombstoneApprovalInboxItem,
16 WorkflowTaskTombstoneApprovalInboxLimit, WorkflowTaskTombstoneApprovalLease,
17 WorkflowTaskTombstoneApprovalWindow, WorkflowTaskTombstoneCursor,
18 WorkflowTaskTombstoneExportReceipt, WorkflowTaskTombstoneGovernanceStore,
19 WorkflowTaskTombstoneLimit, WorkflowTaskTombstonePurgeEvidence, WorkflowTaskTombstonePurgeId,
20 WorkflowTaskTombstonePurgeIntent, WorkflowTaskTombstonePurgeLimit,
21 WorkflowTaskTombstoneRejectionReason, WorkflowTaskTombstoneRetention, WorkflowTenantId,
22};
23
24#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
26#[non_exhaustive]
27pub enum WorkflowTaskGovernancePermission {
28 PlaceHold,
30 ReleaseHold,
32 Export,
34 PreparePurge,
36 ApprovePurge,
38 ReadApprovalInbox,
40 ClaimPurgeApproval,
42 RejectPurge,
44 ExecutePurge,
46 ReadEvidence,
48}
49
50#[derive(Clone, Copy, Debug, Eq, PartialEq)]
52#[non_exhaustive]
53pub enum WorkflowTaskGovernanceOutcome {
54 Succeeded,
56 Denied,
58 AuthorizationError,
60 StoreError,
62 ArchiveError,
64}
65
66pub trait WorkflowTaskGovernanceObserver: Send + Sync {
68 fn observe(
70 &self,
71 permission: WorkflowTaskGovernancePermission,
72 outcome: WorkflowTaskGovernanceOutcome,
73 );
74}
75
76#[derive(Debug, Default)]
77struct NoopWorkflowTaskGovernanceObserver;
78
79impl WorkflowTaskGovernanceObserver for NoopWorkflowTaskGovernanceObserver {
80 fn observe(
81 &self,
82 _permission: WorkflowTaskGovernancePermission,
83 _outcome: WorkflowTaskGovernanceOutcome,
84 ) {
85 }
86}
87
88#[derive(Clone, Debug, Error, Eq, PartialEq)]
90#[error("Task tombstone governance authorization failed: {message}")]
91pub struct WorkflowTaskGovernanceAuthorizationError {
92 message: String,
93}
94
95impl WorkflowTaskGovernanceAuthorizationError {
96 pub fn new(message: impl Into<String>) -> Self {
98 let mut message = message.into();
99 message.truncate(512);
100 Self { message }
101 }
102}
103
104pub type WorkflowTaskGovernanceAuthorizationFuture<'a> = Pin<
106 Box<dyn Future<Output = Result<bool, WorkflowTaskGovernanceAuthorizationError>> + Send + 'a>,
107>;
108
109pub trait WorkflowTaskGovernanceAuthorizer: Send + Sync {
111 fn authorize(
113 &self,
114 principal: &WorkerId,
115 tenant_id: &WorkflowTenantId,
116 permission: WorkflowTaskGovernancePermission,
117 ) -> WorkflowTaskGovernanceAuthorizationFuture<'_>;
118}
119
120#[derive(Clone, Debug, Default)]
122pub struct StaticWorkflowTaskGovernanceAuthorizer {
123 grants: BTreeMap<(WorkerId, WorkflowTenantId), BTreeSet<WorkflowTaskGovernancePermission>>,
124}
125
126impl StaticWorkflowTaskGovernanceAuthorizer {
127 pub fn new() -> Self {
129 Self::default()
130 }
131
132 #[must_use]
134 pub fn with_grant(
135 mut self,
136 principal: WorkerId,
137 tenant_id: WorkflowTenantId,
138 permission: WorkflowTaskGovernancePermission,
139 ) -> Self {
140 self.grants
141 .entry((principal, tenant_id))
142 .or_default()
143 .insert(permission);
144 self
145 }
146}
147
148impl WorkflowTaskGovernanceAuthorizer for StaticWorkflowTaskGovernanceAuthorizer {
149 fn authorize(
150 &self,
151 principal: &WorkerId,
152 tenant_id: &WorkflowTenantId,
153 permission: WorkflowTaskGovernancePermission,
154 ) -> WorkflowTaskGovernanceAuthorizationFuture<'_> {
155 let allowed = self
156 .grants
157 .get(&(principal.clone(), tenant_id.clone()))
158 .is_some_and(|grants| grants.contains(&permission));
159 Box::pin(async move { Ok(allowed) })
160 }
161}
162
163#[derive(Clone, Copy, Debug, Eq, PartialEq)]
165#[non_exhaustive]
166pub enum WorkflowTaskTombstoneArchiveErrorKind {
167 Configuration,
169 Authorization,
171 Timeout,
173 Unavailable,
175 Integrity,
177 Ambiguous,
179 Other,
181}
182
183#[derive(Clone, Debug, Error, Eq, PartialEq)]
185#[error("Task tombstone archive failed ({kind:?}): {message}")]
186pub struct WorkflowTaskTombstoneArchiveError {
187 kind: WorkflowTaskTombstoneArchiveErrorKind,
188 message: String,
189}
190
191impl WorkflowTaskTombstoneArchiveError {
192 pub fn new(message: impl Into<String>) -> Self {
194 Self::with_kind(WorkflowTaskTombstoneArchiveErrorKind::Other, message)
195 }
196
197 pub fn with_kind(
199 kind: WorkflowTaskTombstoneArchiveErrorKind,
200 message: impl Into<String>,
201 ) -> Self {
202 let mut message = message.into();
203 message.truncate(512);
204 Self { kind, message }
205 }
206
207 pub const fn kind(&self) -> WorkflowTaskTombstoneArchiveErrorKind {
209 self.kind
210 }
211}
212
213#[derive(Clone, Debug, Eq, Ord, PartialEq, PartialOrd)]
215pub struct WorkflowTaskTombstoneArchiveBatchId(String);
216
217impl WorkflowTaskTombstoneArchiveBatchId {
218 pub fn parse(value: impl Into<String>) -> Result<Self, WorkflowStoreError> {
224 let value = value.into();
225 if value.trim().is_empty() || value.len() > 512 || value.chars().any(char::is_control) {
226 return Err(WorkflowStoreError::new(
227 crate::WorkflowStoreErrorKind::InvalidInput,
228 "Task tombstone archive batch ID must contain 1..=512 printable bytes",
229 ));
230 }
231 Ok(Self(value))
232 }
233
234 fn from_batch(
235 tenant_id: &WorkflowTenantId,
236 first: WorkflowTaskTombstoneCursor,
237 last: WorkflowTaskTombstoneCursor,
238 ) -> Self {
239 Self(format!(
240 "{}:{}:{}",
241 tenant_id.as_str(),
242 first.get(),
243 last.get()
244 ))
245 }
246
247 pub fn as_str(&self) -> &str {
249 &self.0
250 }
251}
252
253#[derive(Clone, Debug, Eq, PartialEq)]
255pub struct WorkflowTaskTombstoneArchiveBatch {
256 pub batch_id: WorkflowTaskTombstoneArchiveBatchId,
258 pub tenant_id: WorkflowTenantId,
260 pub tombstones: Vec<WorkflowTaskTombstone>,
262}
263
264pub type WorkflowTaskTombstoneArchiveFuture<'a> = Pin<
266 Box<
267 dyn Future<
268 Output = Result<
269 WorkflowTaskTombstoneExportReceipt,
270 WorkflowTaskTombstoneArchiveError,
271 >,
272 > + Send
273 + 'a,
274 >,
275>;
276
277pub trait WorkflowTaskTombstoneArchive: Send + Sync {
279 fn archive(
281 &self,
282 batch: WorkflowTaskTombstoneArchiveBatch,
283 ) -> WorkflowTaskTombstoneArchiveFuture<'_>;
284}
285
286#[derive(Clone, Debug, Eq, PartialEq)]
288pub struct WorkflowTaskTombstoneArchiveReport {
289 pub batch_id: Option<WorkflowTaskTombstoneArchiveBatchId>,
291 pub tombstones_exported: u32,
293 pub through: Option<WorkflowTaskTombstoneCursor>,
295 pub receipt: Option<WorkflowTaskTombstoneExportReceipt>,
297}
298
299#[derive(Debug, Error)]
301#[non_exhaustive]
302pub enum WorkflowTaskGovernanceError {
303 #[error("principal is not authorized for {permission:?}")]
305 Denied {
306 permission: WorkflowTaskGovernancePermission,
308 },
309 #[error(transparent)]
311 Authorization(#[from] WorkflowTaskGovernanceAuthorizationError),
312 #[error(transparent)]
314 Store(#[from] WorkflowStoreError),
315 #[error(transparent)]
317 Archive(#[from] WorkflowTaskTombstoneArchiveError),
318 #[error("cleanup lease owner does not match the authenticated principal")]
320 LeasePrincipalMismatch,
321 #[error("approval lease reviewer does not match the authenticated principal")]
323 ApprovalLeasePrincipalMismatch,
324}
325
326pub struct WorkflowTaskGovernanceControlPlane<S, A> {
328 store: Arc<S>,
329 authorizer: Arc<A>,
330 observer: Arc<dyn WorkflowTaskGovernanceObserver>,
331}
332
333impl<S, A> WorkflowTaskGovernanceControlPlane<S, A>
334where
335 S: WorkflowTaskTombstoneGovernanceStore,
336 A: WorkflowTaskGovernanceAuthorizer,
337{
338 pub fn new(store: Arc<S>, authorizer: Arc<A>) -> Self {
340 Self {
341 store,
342 authorizer,
343 observer: Arc::new(NoopWorkflowTaskGovernanceObserver),
344 }
345 }
346
347 #[must_use]
349 pub fn with_observer(mut self, observer: Arc<dyn WorkflowTaskGovernanceObserver>) -> Self {
350 self.observer = observer;
351 self
352 }
353
354 pub async fn place_hold(
360 &self,
361 principal: &WorkerId,
362 tenant_id: WorkflowTenantId,
363 checkpoint_id: CheckpointId,
364 reason: WorkflowTaskLegalHoldReason,
365 ) -> Result<WorkflowTaskLegalHold, WorkflowTaskGovernanceError> {
366 let permission = WorkflowTaskGovernancePermission::PlaceHold;
367 self.authorize(principal, &tenant_id, permission).await?;
368 let result = self
369 .store
370 .place_task_tombstone_hold(tenant_id, checkpoint_id, principal.clone(), reason)
371 .await;
372 self.store_result(permission, result)
373 }
374
375 pub async fn release_hold(
381 &self,
382 principal: &WorkerId,
383 tenant_id: WorkflowTenantId,
384 checkpoint_id: CheckpointId,
385 ) -> Result<WorkflowTaskLegalHold, WorkflowTaskGovernanceError> {
386 let permission = WorkflowTaskGovernancePermission::ReleaseHold;
387 self.authorize(principal, &tenant_id, permission).await?;
388 let result = self
389 .store
390 .release_task_tombstone_hold(tenant_id, checkpoint_id, principal.clone())
391 .await;
392 self.store_result(permission, result)
393 }
394
395 pub async fn export_next_page<R>(
402 &self,
403 principal: &WorkerId,
404 tenant_id: WorkflowTenantId,
405 after: Option<WorkflowTaskTombstoneCursor>,
406 limit: WorkflowTaskTombstoneLimit,
407 archive: &R,
408 ) -> Result<WorkflowTaskTombstoneArchiveReport, WorkflowTaskGovernanceError>
409 where
410 R: WorkflowTaskTombstoneArchive,
411 {
412 let permission = WorkflowTaskGovernancePermission::Export;
413 self.authorize(principal, &tenant_id, permission).await?;
414 let tombstones = match self
415 .store
416 .list_task_tombstones(tenant_id.clone(), after, limit)
417 .await
418 {
419 Ok(tombstones) => tombstones,
420 Err(error) => return Err(self.store_error(permission, error)),
421 };
422 let (Some(first), Some(last)) = (tombstones.first(), tombstones.last()) else {
423 self.observer
424 .observe(permission, WorkflowTaskGovernanceOutcome::Succeeded);
425 return Ok(WorkflowTaskTombstoneArchiveReport {
426 batch_id: None,
427 tombstones_exported: 0,
428 through: None,
429 receipt: None,
430 });
431 };
432 let batch_id =
433 WorkflowTaskTombstoneArchiveBatchId::from_batch(&tenant_id, first.cursor, last.cursor);
434 let through = last.cursor;
435 let count = u32::try_from(tombstones.len()).unwrap_or(u32::MAX);
436 let receipt = match archive
437 .archive(WorkflowTaskTombstoneArchiveBatch {
438 batch_id: batch_id.clone(),
439 tenant_id: tenant_id.clone(),
440 tombstones,
441 })
442 .await
443 {
444 Ok(receipt) => receipt,
445 Err(error) => {
446 self.observer
447 .observe(permission, WorkflowTaskGovernanceOutcome::ArchiveError);
448 return Err(error.into());
449 }
450 };
451 if let Err(error) = self
452 .store
453 .confirm_task_tombstone_export(tenant_id, through, receipt.clone(), principal.clone())
454 .await
455 {
456 return Err(self.store_error(permission, error));
457 }
458 self.observer
459 .observe(permission, WorkflowTaskGovernanceOutcome::Succeeded);
460 Ok(WorkflowTaskTombstoneArchiveReport {
461 batch_id: Some(batch_id),
462 tombstones_exported: count,
463 through: Some(through),
464 receipt: Some(receipt),
465 })
466 }
467
468 pub async fn prepare_purge(
475 &self,
476 principal: &WorkerId,
477 lease: WorkflowTaskCleanupLease,
478 retention: WorkflowTaskTombstoneRetention,
479 limit: WorkflowTaskTombstonePurgeLimit,
480 approval_window: WorkflowTaskTombstoneApprovalWindow,
481 ) -> Result<WorkflowTaskTombstonePurgeIntent, WorkflowTaskGovernanceError> {
482 let permission = WorkflowTaskGovernancePermission::PreparePurge;
483 self.require_lease_principal(principal, &lease, permission)?;
484 self.authorize(principal, &lease.tenant_id, permission)
485 .await?;
486 let result = self
487 .store
488 .prepare_task_tombstone_purge(lease, retention, limit, approval_window)
489 .await;
490 self.store_result(permission, result)
491 }
492
493 pub async fn approve_purge(
499 &self,
500 principal: &WorkerId,
501 tenant_id: WorkflowTenantId,
502 purge_id: WorkflowTaskTombstonePurgeId,
503 ) -> Result<WorkflowTaskTombstonePurgeIntent, WorkflowTaskGovernanceError> {
504 let permission = WorkflowTaskGovernancePermission::ApprovePurge;
505 self.authorize(principal, &tenant_id, permission).await?;
506 let result = self
507 .store
508 .approve_task_tombstone_purge(tenant_id, purge_id, principal.clone())
509 .await;
510 self.store_result(permission, result)
511 }
512
513 pub async fn list_purge_approvals(
519 &self,
520 principal: &WorkerId,
521 tenant_id: WorkflowTenantId,
522 limit: WorkflowTaskTombstoneApprovalInboxLimit,
523 ) -> Result<Vec<WorkflowTaskTombstoneApprovalInboxItem>, WorkflowTaskGovernanceError> {
524 let permission = WorkflowTaskGovernancePermission::ReadApprovalInbox;
525 self.authorize(principal, &tenant_id, permission).await?;
526 let result = self
527 .store
528 .list_task_tombstone_purge_approvals(tenant_id, limit)
529 .await;
530 self.store_result(permission, result)
531 }
532
533 pub async fn claim_purge_approval(
539 &self,
540 principal: &WorkerId,
541 tenant_id: WorkflowTenantId,
542 lease: LeaseDuration,
543 ) -> Result<Option<WorkflowTaskTombstoneApprovalLease>, WorkflowTaskGovernanceError> {
544 let permission = WorkflowTaskGovernancePermission::ClaimPurgeApproval;
545 self.authorize(principal, &tenant_id, permission).await?;
546 let result = self
547 .store
548 .claim_task_tombstone_purge_approval(tenant_id, principal.clone(), lease)
549 .await;
550 self.store_result(permission, result)
551 }
552
553 pub async fn approve_claimed_purge(
559 &self,
560 principal: &WorkerId,
561 lease: WorkflowTaskTombstoneApprovalLease,
562 ) -> Result<WorkflowTaskTombstonePurgeIntent, WorkflowTaskGovernanceError> {
563 let permission = WorkflowTaskGovernancePermission::ApprovePurge;
564 self.require_approval_principal(principal, &lease, permission)?;
565 self.authorize(principal, &lease.tenant_id, permission)
566 .await?;
567 let result = self.store.approve_claimed_task_tombstone_purge(lease).await;
568 self.store_result(permission, result)
569 }
570
571 pub async fn reject_claimed_purge(
577 &self,
578 principal: &WorkerId,
579 lease: WorkflowTaskTombstoneApprovalLease,
580 reason: WorkflowTaskTombstoneRejectionReason,
581 ) -> Result<WorkflowTaskTombstoneApprovalInboxItem, WorkflowTaskGovernanceError> {
582 let permission = WorkflowTaskGovernancePermission::RejectPurge;
583 self.require_approval_principal(principal, &lease, permission)?;
584 self.authorize(principal, &lease.tenant_id, permission)
585 .await?;
586 let result = self
587 .store
588 .reject_claimed_task_tombstone_purge(lease, reason)
589 .await;
590 self.store_result(permission, result)
591 }
592
593 pub async fn execute_purge(
600 &self,
601 principal: &WorkerId,
602 lease: WorkflowTaskCleanupLease,
603 purge_id: WorkflowTaskTombstonePurgeId,
604 ) -> Result<WorkflowTaskTombstonePurgeEvidence, WorkflowTaskGovernanceError> {
605 let permission = WorkflowTaskGovernancePermission::ExecutePurge;
606 self.require_lease_principal(principal, &lease, permission)?;
607 self.authorize(principal, &lease.tenant_id, permission)
608 .await?;
609 let result = self
610 .store
611 .execute_task_tombstone_purge(lease, purge_id)
612 .await;
613 self.store_result(permission, result)
614 }
615
616 pub async fn get_evidence(
622 &self,
623 principal: &WorkerId,
624 tenant_id: WorkflowTenantId,
625 purge_id: WorkflowTaskTombstonePurgeId,
626 ) -> Result<Option<WorkflowTaskTombstonePurgeEvidence>, WorkflowTaskGovernanceError> {
627 let permission = WorkflowTaskGovernancePermission::ReadEvidence;
628 self.authorize(principal, &tenant_id, permission).await?;
629 let result = self
630 .store
631 .get_task_tombstone_purge_evidence(tenant_id, purge_id)
632 .await;
633 self.store_result(permission, result)
634 }
635
636 async fn authorize(
637 &self,
638 principal: &WorkerId,
639 tenant_id: &WorkflowTenantId,
640 permission: WorkflowTaskGovernancePermission,
641 ) -> Result<(), WorkflowTaskGovernanceError> {
642 match self
643 .authorizer
644 .authorize(principal, tenant_id, permission)
645 .await
646 {
647 Ok(true) => Ok(()),
648 Ok(false) => {
649 self.observer
650 .observe(permission, WorkflowTaskGovernanceOutcome::Denied);
651 Err(WorkflowTaskGovernanceError::Denied { permission })
652 }
653 Err(error) => {
654 self.observer.observe(
655 permission,
656 WorkflowTaskGovernanceOutcome::AuthorizationError,
657 );
658 Err(error.into())
659 }
660 }
661 }
662
663 fn require_lease_principal(
664 &self,
665 principal: &WorkerId,
666 lease: &WorkflowTaskCleanupLease,
667 permission: WorkflowTaskGovernancePermission,
668 ) -> Result<(), WorkflowTaskGovernanceError> {
669 if lease.owner == *principal {
670 Ok(())
671 } else {
672 self.observer
673 .observe(permission, WorkflowTaskGovernanceOutcome::Denied);
674 Err(WorkflowTaskGovernanceError::LeasePrincipalMismatch)
675 }
676 }
677
678 fn require_approval_principal(
679 &self,
680 principal: &WorkerId,
681 lease: &WorkflowTaskTombstoneApprovalLease,
682 permission: WorkflowTaskGovernancePermission,
683 ) -> Result<(), WorkflowTaskGovernanceError> {
684 if lease.reviewer == *principal {
685 Ok(())
686 } else {
687 self.observer
688 .observe(permission, WorkflowTaskGovernanceOutcome::Denied);
689 Err(WorkflowTaskGovernanceError::ApprovalLeasePrincipalMismatch)
690 }
691 }
692
693 fn store_result<T>(
694 &self,
695 permission: WorkflowTaskGovernancePermission,
696 result: Result<T, WorkflowStoreError>,
697 ) -> Result<T, WorkflowTaskGovernanceError> {
698 match result {
699 Ok(value) => {
700 self.observer
701 .observe(permission, WorkflowTaskGovernanceOutcome::Succeeded);
702 Ok(value)
703 }
704 Err(error) => Err(self.store_error(permission, error)),
705 }
706 }
707
708 fn store_error(
709 &self,
710 permission: WorkflowTaskGovernancePermission,
711 error: WorkflowStoreError,
712 ) -> WorkflowTaskGovernanceError {
713 self.observer
714 .observe(permission, WorkflowTaskGovernanceOutcome::StoreError);
715 error.into()
716 }
717}
718
719impl<S, A> std::fmt::Debug for WorkflowTaskGovernanceControlPlane<S, A> {
720 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
721 formatter
722 .debug_struct("WorkflowTaskGovernanceControlPlane")
723 .field("store", &"<governance-store>")
724 .field("authorizer", &"<governance-authorizer>")
725 .field("observer", &"<governance-observer>")
726 .finish()
727 }
728}
729
730#[cfg(test)]
731mod tests {
732 use super::{WorkflowTaskTombstoneArchiveError, WorkflowTaskTombstoneArchiveErrorKind};
733
734 #[test]
735 fn archive_error_preserves_kind_and_bounds_safe_message() {
736 let error = WorkflowTaskTombstoneArchiveError::with_kind(
737 WorkflowTaskTombstoneArchiveErrorKind::Ambiguous,
738 "x".repeat(1_024),
739 );
740
741 assert_eq!(
742 error.kind(),
743 WorkflowTaskTombstoneArchiveErrorKind::Ambiguous
744 );
745 assert!(error.to_string().len() < 600);
746 assert_eq!(
747 WorkflowTaskTombstoneArchiveError::new("fallback").kind(),
748 WorkflowTaskTombstoneArchiveErrorKind::Other
749 );
750 }
751}