From 6e0fcf6b3b78f88bf8a4335edacb0b6ed2c447d6 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 09:29:36 +0200 Subject: [PATCH] Partition workflow execution by tenant --- .../backend/instance_service.py | 23 +++++--- .../backend/manifest.py | 2 +- .../backend/triggers.py | 56 +++++++++++++------ 3 files changed, 55 insertions(+), 26 deletions(-) diff --git a/src/govoplan_workflow_engine/backend/instance_service.py b/src/govoplan_workflow_engine/backend/instance_service.py index cff2c8f..e219341 100644 --- a/src/govoplan_workflow_engine/backend/instance_service.py +++ b/src/govoplan_workflow_engine/backend/instance_service.py @@ -779,15 +779,19 @@ def reconcile_pending_instances( session: Session, *, registry: object | None, + tenant_id: str | None = None, limit: int = 50, ) -> dict[str, object]: + clauses = [ + WorkflowInstance.status.in_(("running", "waiting")), + WorkflowInstance.current_step_id.is_not(None), + ] + if tenant_id: + clauses.append(WorkflowInstance.tenant_id == tenant_id) instances = list( session.scalars( select(WorkflowInstance) - .where( - WorkflowInstance.status.in_(("running", "waiting")), - WorkflowInstance.current_step_id.is_not(None), - ) + .where(*clauses) .order_by(WorkflowInstance.updated_at, WorkflowInstance.id) .limit(max(1, min(int(limit), 200))) .with_for_update(skip_locked=True) @@ -2875,19 +2879,20 @@ def _notify_handoff( class SqlWorkflowRuntimeWorker: def __init__(self, *, registry: object | None = None) -> None: self._registry = registry - self._standards_reconciled = False + self._standards_reconciled_tenants: set[str | None] = set() def reconcile_pending( self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> Mapping[str, object]: if not isinstance(session, Session): raise TypeError("Workflow reconciliation requires a Session.") standards: Mapping[str, object] | None = None - if not self._standards_reconciled: + if tenant_id not in self._standards_reconciled_tenants: from govoplan_workflow_engine.backend.contributions import ( reconcile_workflow_definition_contributions, ) @@ -2895,11 +2900,14 @@ class SqlWorkflowRuntimeWorker: standards = reconcile_workflow_definition_contributions( session, registry=self._registry, + tenant_ids=((tenant_id,) if tenant_id else ()), ) - self._standards_reconciled = int(standards.get("blocked") or 0) == 0 + if int(standards.get("blocked") or 0) == 0: + self._standards_reconciled_tenants.add(tenant_id) runtime = reconcile_pending_instances( session, registry=self._registry, + tenant_id=tenant_id, limit=limit, ) from govoplan_workflow_engine.backend.triggers import dispatch_due_work @@ -2907,6 +2915,7 @@ class SqlWorkflowRuntimeWorker: triggers = dispatch_due_work( session, registry=self._registry, + tenant_id=tenant_id, now=now, limit=limit, ) diff --git a/src/govoplan_workflow_engine/backend/manifest.py b/src/govoplan_workflow_engine/backend/manifest.py index 715e34d..87cd9ca 100644 --- a/src/govoplan_workflow_engine/backend/manifest.py +++ b/src/govoplan_workflow_engine/backend/manifest.py @@ -406,7 +406,7 @@ manifest = ModuleManifest( "outcome and disables Retry. Operators inspect the provider, record " "evidence, and choose Effect confirmed to continue without replay or " "Effect absent to enable a deliberate retry. Instance workers, " - "trigger deliveries, and timers use distributed fences. Linked " + "trigger deliveries, and timers use distributed fences and are partitioned by tenant module entitlement before state is claimed. Disabling Workflow Engine preserves accepted instances, waits, and trigger deliveries for operator resolution. Linked " "Dataflow recovery remains blocked until its result is conclusive." ), layer="available", diff --git a/src/govoplan_workflow_engine/backend/triggers.py b/src/govoplan_workflow_engine/backend/triggers.py index 70b2c10..e69ce2f 100644 --- a/src/govoplan_workflow_engine/backend/triggers.py +++ b/src/govoplan_workflow_engine/backend/triggers.py @@ -351,16 +351,25 @@ def dispatch_due_work( session: Session, *, registry: object | None, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> dict[str, int]: current = _as_utc(now or utcnow()) bounded = max(1, min(int(limit), 200)) - scheduled = _queue_due_schedules(session, now=current, limit=bounded) + scheduled = _queue_due_schedules( + session, + now=current, + limit=bounded, + tenant_id=tenant_id, + ) + delivery_clauses = [WorkflowTriggerDelivery.status == "queued"] + if tenant_id: + delivery_clauses.append(WorkflowTriggerDelivery.tenant_id == tenant_id) deliveries = list( session.scalars( select(WorkflowTriggerDelivery) - .where(WorkflowTriggerDelivery.status == "queued") + .where(*delivery_clauses) .order_by( WorkflowTriggerDelivery.created_at, WorkflowTriggerDelivery.id, @@ -396,6 +405,7 @@ def dispatch_due_work( registry=registry, now=current, limit=bounded, + tenant_id=tenant_id, ) session.flush() return { @@ -416,6 +426,7 @@ class SqlWorkflowTriggerDispatcher: self, session: object, *, + tenant_id: str | None = None, now: datetime | None = None, limit: int = 50, ) -> Mapping[str, object]: @@ -424,6 +435,7 @@ class SqlWorkflowTriggerDispatcher: return dispatch_due_work( session, registry=self._registry, + tenant_id=tenant_id, now=now, limit=limit, ) @@ -444,16 +456,20 @@ def _queue_due_schedules( *, now: datetime, limit: int, + tenant_id: str | None = None, ) -> int: + clauses = [ + WorkflowTrigger.kind == "schedule", + WorkflowTrigger.status == "enabled", + WorkflowTrigger.next_fire_at.is_not(None), + WorkflowTrigger.next_fire_at <= now, + ] + if tenant_id: + clauses.append(WorkflowTrigger.tenant_id == tenant_id) triggers = list( session.scalars( select(WorkflowTrigger) - .where( - WorkflowTrigger.kind == "schedule", - WorkflowTrigger.status == "enabled", - WorkflowTrigger.next_fire_at.is_not(None), - WorkflowTrigger.next_fire_at <= now, - ) + .where(*clauses) .order_by(WorkflowTrigger.next_fire_at, WorkflowTrigger.id) .limit(limit) .with_for_update(skip_locked=True) @@ -606,20 +622,24 @@ def _dispatch_waits( registry: object | None, now: datetime, limit: int, + tenant_id: str | None = None, ) -> dict[str, int]: + clauses = [ + or_( + WorkflowWaitState.status == "triggered", + and_( + WorkflowWaitState.status == "waiting", + WorkflowWaitState.due_at.is_not(None), + WorkflowWaitState.due_at <= now, + ), + ) + ] + if tenant_id: + clauses.append(WorkflowWaitState.tenant_id == tenant_id) states = list( session.scalars( select(WorkflowWaitState) - .where( - or_( - WorkflowWaitState.status == "triggered", - and_( - WorkflowWaitState.status == "waiting", - WorkflowWaitState.due_at.is_not(None), - WorkflowWaitState.due_at <= now, - ), - ) - ) + .where(*clauses) .order_by(WorkflowWaitState.updated_at, WorkflowWaitState.id) .limit(limit) .with_for_update(skip_locked=True)