-- broker.broker_get -- Claims the next eligible job from a queue: pending, available (backoff -- elapsed), no incomplete dependency, and visible under the caller's RLS -- tenant. Grants a lease (lease_token) that broker_run must present back. -- Returns: p_retval (0=success, >0=infra error), p_errmsg, p_job_id, p_lease_token. CREATE OR REPLACE FUNCTION broker.broker_get( p_queue_number INTEGER, p_instance_id BIGINT DEFAULT NULL, p_lease_seconds INTEGER DEFAULT 60, OUT p_retval INTEGER, OUT p_errmsg TEXT, OUT p_job_id BIGINT, OUT p_lease_token UUID ) RETURNS RECORD LANGUAGE plpgsql AS $$ DECLARE v_job_id BIGINT; v_lease_token UUID; BEGIN p_retval := 0; p_errmsg := ''; p_job_id := NULL; p_lease_token := NULL; IF p_queue_number IS NULL OR p_queue_number <= 0 THEN p_retval := 1; p_errmsg := 'Invalid queue number'; RETURN; END IF; IF p_lease_seconds IS NULL OR p_lease_seconds <= 0 THEN p_lease_seconds := 60; END IF; SELECT candidate.id_broker_jobs INTO v_job_id FROM broker.broker_jobs candidate WHERE candidate.job_queue = p_queue_number AND candidate.complete_status = 0 AND candidate.available_at <= NOW() AND NOT EXISTS ( SELECT 1 FROM broker.broker_job_dependency d JOIN broker.broker_jobs dep ON dep.id_broker_jobs = d.depends_on_job_id WHERE d.job_id = candidate.id_broker_jobs AND dep.complete_status <> 2 ) ORDER BY candidate.job_priority DESC, candidate.created_at ASC, candidate.id_broker_jobs ASC LIMIT 1 FOR UPDATE SKIP LOCKED; IF NOT FOUND THEN RETURN; END IF; v_lease_token := gen_random_uuid(); UPDATE broker.broker_jobs SET complete_status = 1, -- running started_at = NOW(), rid_broker_queueinstance = p_instance_id, attempt_count = attempt_count + 1, lease_token = v_lease_token, leased_at = NOW(), lease_expires_at = NOW() + make_interval(secs => p_lease_seconds), updated_at = NOW() WHERE id_broker_jobs = v_job_id; p_job_id := v_job_id; p_lease_token := v_lease_token; EXCEPTION WHEN OTHERS THEN p_retval := 2; p_errmsg := SQLERRM; RAISE WARNING 'broker_get error: %', SQLERRM; END; $$; COMMENT ON FUNCTION broker.broker_get IS 'Claims the next eligible job from a queue and grants a lease';