58 lines
1.7 KiB
PL/PgSQL
58 lines
1.7 KiB
PL/PgSQL
BEGIN;
|
|
|
|
-- A user-authorized immediate dispatch can encounter a transactionally
|
|
-- enqueued system attempt. That path already owns the event/attempt row before
|
|
-- this trigger runs, while the drain owns the global rate lock before rows.
|
|
-- Never wait for the global lock here: an immediate serialization failure
|
|
-- rolls the user statement back and leaves the durable pending row for drain.
|
|
CREATE OR REPLACE FUNCTION public.enforce_system_push_claim_rate()
|
|
RETURNS trigger
|
|
LANGUAGE plpgsql
|
|
SECURITY DEFINER
|
|
SET search_path = ''
|
|
AS $$
|
|
BEGIN
|
|
IF NEW.actor_kind <> 'system'
|
|
OR NEW.status <> 'processing'
|
|
OR NEW.dispatch_lease_token IS NULL THEN
|
|
RETURN NEW;
|
|
END IF;
|
|
IF TG_OP = 'UPDATE'
|
|
AND OLD.dispatch_lease_token IS NOT DISTINCT FROM NEW.dispatch_lease_token THEN
|
|
RETURN NEW;
|
|
END IF;
|
|
|
|
IF NOT pg_try_advisory_xact_lock(
|
|
hashtextextended('system-push-global', 73050)
|
|
) THEN
|
|
RAISE EXCEPTION 'push_claim_busy' USING ERRCODE = '40001';
|
|
END IF;
|
|
|
|
DELETE FROM public.push_dispatch_claim_events
|
|
WHERE claimed_at < now() - interval '2 days';
|
|
|
|
IF (
|
|
SELECT count(*)
|
|
FROM public.push_dispatch_claim_events
|
|
WHERE claimed_at > now() - interval '1 minute'
|
|
) >= 500 OR (
|
|
SELECT count(*)
|
|
FROM public.push_dispatch_claim_events
|
|
WHERE claimed_at > now() - interval '24 hours'
|
|
) >= 10000 THEN
|
|
RAISE EXCEPTION 'push_rate_limited' USING ERRCODE = '54000';
|
|
END IF;
|
|
|
|
INSERT INTO public.push_dispatch_claim_events (
|
|
attempt_id, dispatch_lease_token
|
|
) VALUES (
|
|
NEW.id, NEW.dispatch_lease_token
|
|
);
|
|
RETURN NEW;
|
|
END;
|
|
$$;
|
|
|
|
REVOKE ALL ON FUNCTION public.enforce_system_push_claim_rate()
|
|
FROM PUBLIC, anon, authenticated;
|
|
|
|
COMMIT;
|