DataOps-platform/migrations/versions/20260817_518_tenant_outbox_attempt_counter.py

"""Count each durable worker lease so failure retries are bounded."""
from alembic import op

revision="20260817_518"
down_revision="20260817_517"
branch_labels=None
depends_on=None
def upgrade() -> None:
 op.execute("""
 CREATE OR REPLACE FUNCTION public.tenant_control_worker_claim(p_payload jsonb) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $f$
 DECLARE v uuid; w text; s integer; r public.tenant_control_outbox%ROWTYPE; t text;
 BEGIN IF NOT pg_has_role(session_user,'dataops_tenant_control','MEMBER') THEN RAISE EXCEPTION 'tenant_control_identity_required'; END IF; v:=(p_payload->>'outbox_uid')::uuid; w:=p_payload->>'worker_id'; s:=(p_payload->>'lease_seconds')::integer; SELECT tenant_id INTO t FROM public.tenant_control_outbox_scope_lookup WHERE outbox_uid=v; IF NOT FOUND THEN RAISE EXCEPTION 'tenant_outbox_missing'; END IF; PERFORM set_config('dataops.tenant_id',t,true); SELECT * INTO r FROM public.tenant_control_outbox WHERE outbox_uid=v FOR UPDATE; IF NOT FOUND OR r.state='completed' OR r.state='failed' OR (r.state='leased' AND r.lease_expires_at>clock_timestamp()) OR r.attempt_count>=3 OR (SELECT state FROM public.tenants WHERE tenant_id=t)<>'active' THEN RAISE EXCEPTION 'tenant_worker_scope_denied'; END IF; UPDATE public.tenant_control_outbox SET state='leased',worker_id=w,leased_at=clock_timestamp(),lease_expires_at=clock_timestamp()+make_interval(secs=>s),lease_fence=lease_fence+1,attempt_count=attempt_count+1 WHERE outbox_uid=v RETURNING * INTO r; RETURN jsonb_build_object('outbox_uid',r.outbox_uid,'tenant_id',r.tenant_id,'event_type',r.event_type,'payload_digest',r.payload_digest,'lease_fence',r.lease_fence,'lease_expires_at',r.lease_expires_at); END;$f$;
 ALTER FUNCTION public.tenant_control_worker_claim(jsonb) OWNER TO dataops_tenant_foundation_owner; REVOKE ALL ON FUNCTION public.tenant_control_worker_claim(jsonb) FROM PUBLIC,dataops_app_runtime; GRANT EXECUTE ON FUNCTION public.tenant_control_worker_claim(jsonb) TO dataops_tenant_control;
 """)
def downgrade() -> None: pass